/*
 * brin_minmax_multi.c
 *		BRIN 的多重最小/最大操作类的实现
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * 实现了一种变体的最小最大操作类，其中摘要由多个较小的区间组成。这使我们能够处理
 * 异常值，这通常使简单的最小最大操作类效率低下。
 *
 * 例如，考虑具有简单最小最大区间 [1000,2000] 的页面范围，并假设一个新行以值 1000000
 * 插入到该范围中。由于该区间变为 [1000,1000000]。也就是说，最小最大区间变宽了 1000
 * 倍，并且在 2001 和 1000000 之间将无法有效消除扫描键。
 *
 * 使用最小最大多重操作类，我们最开始可能拥有区间 [1000,2000]，但在添加新行后，我们开始
 * 将其跟踪为两个区间：
 *
 *   [1000,2000] 和 [1000000,1000000]
 *
 * 这使我们在扫描键落在 2000 和 1000000 之间的间隙时仍然能够消除页面范围，在简单的最小
 * 最大操作类效率低下的情况下依然有效。
 *
 * 每个页面范围跟踪的区间数量是相对灵活的。限制的是每个页面范围的值的数量，当前限制为
 * 32（见 values_per_range 选项）。合并区间（最小值和最大值相等）作为单个值存储，而
 * 正常的区间则需要两个值。
 *
 * 当值的数量过多（通过向摘要中添加新值），我们会合并一些区间以释放空间给更多的值。这
 * 是以贪婪的方式完成的 - 我们简单选择两个最近的区间，合并它们，并重复此过程，直到存
 * 储的值的数量变得足够低（低于最大值的 50%），但这大多是任意阈值，且可轻易更改。
 *
 * 为了选择最近的区间，我们使用“距离”支持过程，该过程测量两个范围之间的空间（即区间的
 * 长度）。计算出的值可能是近似值 - 在最坏的情况下，我们将在该步骤合并两个稍微不太优
 * 化的区间，但索引应该仍然产生正确的结果。
 *
 * 压缩（减少值的数量）相当昂贵，因为它需要调用距离函数、排序等。因此，在构建摘要时，
 * 我们使用显著更大的缓冲区，并且仅在最后强制执行确切限制。这提高了性能，同时也有助于
 * 构建更好的区间（由于其贪婪的方法）。
 *
 *
 * IDENTIFICATION
 *	  src/backend/access/brin/brin_minmax_multi.c
 */
#include "postgres.h"

/* 需要 PGSQL_AF_INET */
#include <sys/socket.h>

#include "access/genam.h"
#include "access/brin.h"
#include "access/brin_internal.h"
#include "access/brin_tuple.h"
#include "access/reloptions.h"
#include "access/stratnum.h"
#include "access/htup_details.h"
#include "catalog/pg_type.h"
#include "catalog/pg_am.h"
#include "catalog/pg_amop.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/date.h"
#include "utils/datum.h"
#include "utils/float.h"
#include "utils/inet.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/numeric.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/syscache.h"
#include "utils/timestamp.h"
#include "utils/uuid.h"

/*
 * 额外的 SQL 层支持函数
 *
 * 过程编号不得使用保留给 BRIN 本身的值；请参见
 * brin_internal.h。
 */
#define		MINMAX_MAX_PROCNUMS		1	/* 我们需要的最大支持过程 */
#define		PROCNUM_DISTANCE		11	/* 必需，值之间的距离 */

/*
 * 从 procnum 中减去此值以获得 MinmaxMultiOpaque 数组中的索引
 * （必须等于私有 procnums 的最小值）。
 */
#define		PROCNUM_BASE			11

/*
 * 插入缓冲区的大小 - 我们使用指定值数量的 10 倍
 * 在 reloption 中，但我们将其限制为 8192，以免过大。当
 * 缓冲区满时，我们将值的数量减少一半。
 */
#define		MINMAX_BUFFER_FACTOR			10
#define		MINMAX_BUFFER_MIN				256
#define		MINMAX_BUFFER_MAX				8192
#define		MINMAX_BUFFER_LOAD_FACTOR		0.5

typedef struct MinmaxMultiOpaque
{
	FmgrInfo	extra_procinfos[MINMAX_MAX_PROCNUMS];
	bool		extra_proc_missing[MINMAX_MAX_PROCNUMS];
	Oid			cached_subtype;
	FmgrInfo	strategy_procinfos[BTMaxStrategyNumber];
} MinmaxMultiOpaque;

/*
 * BRIN 的 minmax reloptions 的存储类型
 */
typedef struct MinMaxMultiOptions
{
	int32		vl_len_;		/* varlena 头部（请勿直接触摸！） */
	int			valuesPerRange; /* 每个范围的值数量 */
} MinMaxMultiOptions;

#define MINMAX_MULTI_DEFAULT_VALUES_PER_PAGE		32

#define MinMaxMultiGetValuesPerRange(opts) \
		((opts) && (((MinMaxMultiOptions *) (opts))->valuesPerRange != 0) ? \
		 ((MinMaxMultiOptions *) (opts))->valuesPerRange : \
		 MINMAX_MULTI_DEFAULT_VALUES_PER_PAGE)

#define SAMESIGN(a,b) (((a) < 0) == ((b) < 0))

/*
 * minmax-multi 索引的摘要有两种表示 - 方便处理的范围和
 * 用于在 bytea 值中存储的 SerializedRanges。
 *
 * Ranges 结构将边界值存储在一个数组中，但我们
 * 将常规和单点范围区别对待以节省空间。对于
 * 常规范围（具有不同边界值），我们必须存储范围的
 * 下限和上限，而对于“单点范围”
 * 我们只需要存储一个值。
 *
 * 'values' 数组首先存储常规范围的边界值（有 2*nranges 个值需要存储），
 * 然后是单点范围的 nvalues 边界值。也就是说，我们在数组中有
 * (2*nranges + nvalues) 个边界值。
 *
 * +-------------------------+----------------------------------+
 * | 范围 (2 * nranges 个) | 单点值 (nvalues 个) |
 * +-------------------------+----------------------------------+
 *
 * 这使我们能够快速添加新值，并存储离群值，而无需
 * 加宽任何现有的范围值。
 *
 * 'nsorted' 表示 values[] 数组中有多少个 'nvalues' 是排序的。
 * 当 nsorted == nvalues 时，所有单点值都是排序的。
 *
 * 我们从不存储超过 maxvalues 的值（由 values_per_range
 * reloption 设置）。如有必要，我们合并一些范围。
 *
 * 为了最小化 palloc 开销，我们始终为
 * maxvalues 元素分配完整数组的空间。只要
 * maxvalues 合理较小（64 看起来不错），这应该没问题，
 * 这得益于 values_per_range reloption 的限制为 256。
 */
typedef struct Ranges
{
	/* 我们经常需要的缓存信息。 */
	Oid			typid;
	Oid			colloid;
	AttrNumber	attno;
	FmgrInfo   *cmp;

	/* (2*nranges + nvalues) <= maxvalues */
	int			nranges;		/* values[] 数组中的范围数量 */
	int			nsorted;		/* 已排序的 nvalues 数量 */
	int			nvalues;		/* values[] 数组中的点值数量 */
	int			maxvalues;		/* values[] 数组中的元素数量 */

	/*
	 * 我们只是将值添加到一个大缓冲区中，没有任何昂贵的步骤
	 * （排序、去重等）。缓冲区是目标值数量的倍数，因此压缩发生的频率较低，
	 * 从而摊销成本。在序列化为
	 * 磁盘表示之前，我们保持实际目标并压缩到请求的值数量。
	 */
	/* 请求的值数量 */
	int			target_maxvalues;

	/* 存储在此范围内的值 - 原始值或范围 */
	Datum		values[FLEXIBLE_ARRAY_MEMBER];
} Ranges;

/*
 * 在磁盘上，摘要作为一个bytea值存储，简单的头部
 * 包含基本元数据，后面是边界值。它有一个varlena
 * 头部，因此可以直接作为varlena处理。
 *
 * 请参见brin_range_serialize/brin_range_deserialize以获取序列化细节。
 */
typedef struct SerializedRanges
{
	/* varlena 头部（请勿直接触摸！） */
	int32		vl_len_;

	/* 存储在数据数组中的值的类型 */
	Oid			typid;

	/* (2*nranges + nvalues) <= maxvalues */
	int			nranges;		/* 数组中的范围数量（已存储） */
	int			nvalues;		/* 数据数组中的值的数量（全部） */
	int			maxvalues;		/* 最大值数量（reloption） */

	/* 包含实际数据 */
	char		data[FLEXIBLE_ARRAY_MEMBER];
} SerializedRanges;

static SerializedRanges *fc_brin_range_serialize(Ranges *fc_range);

static Ranges *fc_brin_range_deserialize(int fc_maxvalues, SerializedRanges *fc_range);


/*
 * 用于表示扩展的范围，以便于合并和组合。
 *
 * 每个扩展的范围本质上是一个区间，由最小/最大
 * 值表示，以及一个标志指示它是否为折叠范围（在这种情况下
 * 最小值和最大值相等）。我们有这个标志来处理按引用的
 * 数据类型 - 我们不能简单地比较数据项，这样可以节省一些
 * 调用特定于类型的比较函数。
 */
typedef struct ExpandedRange
{
	Datum		minval;			/* 下边界 */
	Datum		maxval;			/* 上边界 */
	bool		collapsed;		/* 如果minval==maxval则为真 */
} ExpandedRange;

/*
 * 表示两个范围之间的距离（通过索引识别
 * 一个扩展范围的数组）。
 */
typedef struct DistanceValue
{
	int			index;
	double		value;
} DistanceValue;


/* 支持和策略过程的缓存。 */

static FmgrInfo *fc_minmax_multi_get_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno,
										   uint16 fc_procnum);

static FmgrInfo *fc_minmax_multi_get_strategy_procinfo(BrinDesc *fc_bdesc,
													uint16 fc_attno, Oid fc_subtype,
													uint16 fc_strategynum);

typedef struct compare_context
{
	FmgrInfo   *cmpFn;
	Oid			colloid;
} compare_context;

static int	fc_compare_values(const void *fc_a, const void *fc_b, void *fc_arg);


#ifdef USE_ASSERT_CHECKING
/*
 * 使用cmp函数（应该是BTLessStrategyNumber）检查数组值的顺序是否正确。
 */
static void fc_AssertArrayOrder(FmgrInfo *fc_cmp, Oid fc_colloid, Datum *fc_values, int fc_nvalues)
{
	int			fc_i;
	Datum		fc_lt;

	for (fc_i = 0; fc_i < (fc_nvalues - 1); fc_i++)
	{
		fc_lt = FunctionCall2Coll(fc_cmp, fc_colloid, fc_values[fc_i], fc_values[fc_i + 1]);
		Assert(DatumGetBool(fc_lt));
	}
}
#endif

/*
 * 对范围结构进行全面检查。
 */
static void fc_AssertCheckRanges(Ranges *fc_ranges, FmgrInfo *fc_cmpFn, Oid fc_colloid)
{
#ifdef USE_ASSERT_CHECKING
	int			fc_i;

	/* 一些基本的合理性检查 */
	Assert(fc_ranges->nranges >= 0);
	Assert(fc_ranges->nsorted >= 0);
	Assert(fc_ranges->nvalues >= fc_ranges->nsorted);
	Assert(fc_ranges->maxvalues >= 2 * fc_ranges->nranges + fc_ranges->nvalues);
	Assert(fc_ranges->typid != InvalidOid);

	/*
	 * 首先是范围 - 有2*nranges个边界值，值
	 * 必须严格排序（相等的值意味着范围被
	 * 折叠，应该作为一个点存储）。这也保证了
	 * 范围不重叠。
	 */
	fc_AssertArrayOrder(fc_cmpFn, fc_colloid, fc_ranges->values, 2 * fc_ranges->nranges);

	/* 然后是单点范围（带有nvalues边界值） */
	fc_AssertArrayOrder(fc_cmpFn, fc_colloid, &fc_ranges->values[2 * fc_ranges->nranges],
					 fc_ranges->nsorted);

	/*
	 * 检查没有任何值未被范围覆盖（已排序和未排序）
	 */
	if (fc_ranges->nranges > 0)
	{
		for (fc_i = 0; fc_i < fc_ranges->nvalues; fc_i++)
		{
			Datum		fc_compar;
			int			fc_start,
						fc_end;
			Datum		fc_minvalue = fc_ranges->values[0];
			Datum		fc_maxvalue = fc_ranges->values[2 * fc_ranges->nranges - 1];
			Datum		fc_value = fc_ranges->values[2 * fc_ranges->nranges + fc_i];

			fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_value, fc_minvalue);

			/*
			 * 如果值小于第一范围的下边界
			 * 那么它不可能在任何范围内。
			 */
			if (DatumGetBool(fc_compar))
				continue;

			fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_maxvalue, fc_value);

			/*
			 * 同样，如果值大于最后一个范围的上边界，
			 * 那么它不可能在任何范围内。
			 */
			if (DatumGetBool(fc_compar))
				continue;

			/* 二分搜索范围以查看“值”是否适合其中任何一个 */
			fc_start = 0;			/* 第一个范围 */
			fc_end = fc_ranges->nranges - 1;	/* 最后一个范围 */
			while (true)
			{
				int			fc_midpoint = (fc_start + fc_end) / 2;

				/* 这意味着我们在上一步中耗尽了范围 */
				if (fc_start > fc_end)
					break;

				/* 复制范围的最小/最大值 */
				fc_minvalue = fc_ranges->values[2 * fc_midpoint];
				fc_maxvalue = fc_ranges->values[2 * fc_midpoint + 1];

				/*
				 * 值是否小于minval？如果是，我们将递归到范围数组的左侧。
				 */
				fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_value, fc_minvalue);

				/* 小于此范围内的最小值 */
				if (DatumGetBool(fc_compar))
				{
					fc_end = (fc_midpoint - 1);
					continue;
				}

				/*
				 * 值是否大于minval？如果是，我们将递归到范围数组的右侧。
				 */
				fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_maxvalue, fc_value);

				/* 大于此范围内的最大值 */
				if (DatumGetBool(fc_compar))
				{
					fc_start = (fc_midpoint + 1);
					continue;
				}

				/* 嘿，我们找到了匹配的范围 */
				Assert(false);
			}
		}
	}

	/* 且未排序部分的值不能在已排序部分中 */
	if (fc_ranges->nsorted > 0)
	{
		compare_context fc_cxt;

		fc_cxt.colloid = fc_ranges->colloid;
		fc_cxt.cmpFn = fc_ranges->cmp;

		for (fc_i = fc_ranges->nsorted; fc_i < fc_ranges->nvalues; fc_i++)
		{
			Datum		fc_value = fc_ranges->values[2 * fc_ranges->nranges + fc_i];

			Assert(bsearch_arg(&fc_value, &fc_ranges->values[2 * fc_ranges->nranges],
							   fc_ranges->nsorted, sizeof(Datum),
							   fc_compare_values, (void *) &fc_cxt) == NULL);
		}
	}
#endif
}

/*
 * 检查扩展的范围（通过合并一些范围来减少范围数量时构建的）是否正确排序且不重叠。
 */
static void fc_AssertCheckExpandedRanges(BrinDesc *fc_bdesc, Oid fc_colloid, AttrNumber fc_attno,
						  Form_pg_attribute fc_attr, ExpandedRange *fc_ranges,
						  int fc_nranges)
{
#ifdef USE_ASSERT_CHECKING
	int			fc_i;
	FmgrInfo   *fc_eq;
	FmgrInfo   *fc_lt;

	fc_eq = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
											BTEqualStrategyNumber);

	fc_lt = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
											BTLessStrategyNumber);

	/*
	 * 每个范围应该独立有效，即边界值（下限 <= 上限）。
	 */
	for (fc_i = 0; fc_i < fc_nranges; fc_i++)
	{
		Datum		fc_r;
		Datum		fc_minval = fc_ranges[fc_i].minval;
		Datum		fc_maxval = fc_ranges[fc_i].maxval;

		if (fc_ranges[fc_i].collapsed)	/* 合并：minval == maxval */
			fc_r = FunctionCall2Coll(fc_eq, fc_colloid, fc_minval, fc_maxval);
		else					/* 非合并：minval < maxval */
			fc_r = FunctionCall2Coll(fc_lt, fc_colloid, fc_minval, fc_maxval);

		Assert(DatumGetBool(fc_r));
	}

	/*
	 * 范围应该是有序的且必须不重叠，即连续范围的边界上限 <
	 * 下限。
	 */
	for (fc_i = 0; fc_i < fc_nranges - 1; fc_i++)
	{
		Datum		fc_r;
		Datum		fc_maxval = fc_ranges[fc_i].maxval;
		Datum		fc_minval = fc_ranges[fc_i + 1].minval;

		fc_r = FunctionCall2Coll(fc_lt, fc_colloid, fc_maxval, fc_minval);

		Assert(DatumGetBool(fc_r));
	}
#endif
}


/*
 * minmax_multi_init
 * 		初始化反序列化的范围列表，分配所有内存。
 *
 * 这只是范围的内存表示，因此我们为最大值数量分配足够的空间（以免在范围增长时进行重新分配）。
 */
static Ranges *
fc_minmax_multi_init(int fc_maxvalues)
{
	Size		fc_len;
	Ranges	   *fc_ranges;

	Assert(fc_maxvalues > 0);

	fc_len = offsetof(Ranges, values); /* 固定头部 */
	fc_len += fc_maxvalues * sizeof(Datum);	/* Datum值 */

	fc_ranges = (Ranges *) palloc0(fc_len);

	fc_ranges->maxvalues = fc_maxvalues;

	return fc_ranges;
}


/*
 * range_deduplicate_values
 *		在简单点中去重值的部分。
 *
 * 这被认为是一种更省钱的方法来减少范围的大小。它不触碰范围，只对其他值进行排序——它不调用距离函数，这可能相当昂贵，等等。
 *
 * 我们知道这些值与范围没有重复，因为在添加新值之前我们检查了这一点。已排序部分的值也是如此。
 */
static void fc_range_deduplicate_values(Ranges *fc_range)
{
	int			fc_i,
				fc_n;
	int			fc_start;
	compare_context fc_cxt;

	/*
	 * 如果没有未排序的值，我们就完成了（这可能不会发生，因为我们正在将值添加到未排序部分）。
	 */
	if (fc_range->nsorted == fc_range->nvalues)
		return;

	/* 对值进行排序 */
	fc_cxt.colloid = fc_range->colloid;
	fc_cxt.cmpFn = fc_range->cmp;

	/* 值在范围之后开始（范围总是已排序的） */
	fc_start = 2 * fc_range->nranges;

	/*
	 * XXX 这可能会进行归并排序，以利用数组的前半部分已排序的特性。如果已排序部分很大，可能会更快。
	 */
	qsort_arg(&fc_range->values[fc_start],
			  fc_range->nvalues, sizeof(Datum),
			  fc_compare_values, (void *) &fc_cxt);

	fc_n = 1;
	for (fc_i = 1; fc_i < fc_range->nvalues; fc_i++)
	{
		/* 与前一值相同，因此存储它 */
		if (fc_compare_values(&fc_range->values[fc_start + fc_i - 1],
						   &fc_range->values[fc_start + fc_i],
						   (void *) &fc_cxt) == 0)
			continue;

		fc_range->values[fc_start + fc_n] = fc_range->values[fc_start + fc_i];

		fc_n++;
	}

	/* 现在所有值都已排序 */
	fc_range->nvalues = fc_n;
	fc_range->nsorted = fc_n;

	fc_AssertCheckRanges(fc_range, fc_range->cmp, fc_range->colloid);
}


/*
 * brin_range_serialize
 *	  将内存表示序列化为紧凑的varlena值。
 *
 * 简单复制头部，然后再复制在内存值数组中存储的单个值。
 */
static SerializedRanges *
fc_brin_range_serialize(Ranges *fc_range)
{
	Size		fc_len;
	int			fc_nvalues;
	SerializedRanges *fc_serialized;
	Oid			fc_typid;
	int			fc_typlen;
	bool		fc_typbyval;

	int			fc_i;
	char	   *fc_ptr;

	/* 简单的合理性检查 */
	Assert(fc_range->nranges >= 0);
	Assert(fc_range->nsorted >= 0);
	Assert(fc_range->nvalues >= 0);
	Assert(fc_range->maxvalues > 0);
	Assert(fc_range->target_maxvalues > 0);

	/* 此时范围应压缩到目标大小 */
	Assert(2 * fc_range->nranges + fc_range->nvalues <= fc_range->target_maxvalues);

	Assert(fc_range->target_maxvalues <= fc_range->maxvalues);

	/* 范围边界始终已排序 */
	Assert(fc_range->nvalues >= fc_range->nsorted);

	/* 如果有未排序的部分则去重值 */
	fc_range_deduplicate_values(fc_range);

	/* 查看我们实际有多少个Datum值 */
	fc_nvalues = 2 * fc_range->nranges + fc_range->nvalues;

	fc_typid = fc_range->typid;
	fc_typbyval = get_typbyval(fc_typid);
	fc_typlen = get_typlen(fc_typid);

	/* 头部始终是需要的 */
	fc_len = offsetof(SerializedRanges, data);

	/*
	 * 所需空间取决于数据类型 - 对于固定长度的数据类型
	 * (按值和某些按引用)很简单，只需乘以
	 * (attlen * nvalues) 就可以了。对于可变长度的按引用
	 * 类型，我们需要遍历所有值并计算总长度。
	 */
	if (fc_typlen == -1)			/* varlena */
	{
		int			fc_i;

		for (fc_i = 0; fc_i < fc_nvalues; fc_i++)
		{
			fc_len += VARSIZE_ANY(fc_range->values[fc_i]);
		}
	}
	else if (fc_typlen == -2)		/* cstring */
	{
		int			fc_i;

		for (fc_i = 0; fc_i < fc_nvalues; fc_i++)
		{
			/* 不要忘记包括 null 终止符 ;-) */
			fc_len += strlen(DatumGetCString(fc_range->values[fc_i])) + 1;
		}
	}
	else						/* 固定长度类型（即使是按引用） */
	{
		Assert(fc_typlen > 0);
		fc_len += fc_nvalues * fc_typlen;
	}

	/*
	 * 分配序列化对象，复制基本信息。序列化对象是一个 varlena，
	 * 所以要更新头部。
	 */
	fc_serialized = (SerializedRanges *) palloc0(fc_len);
	SET_VARSIZE(fc_serialized, fc_len);

	fc_serialized->typid = fc_typid;
	fc_serialized->nranges = fc_range->nranges;
	fc_serialized->nvalues = fc_range->nvalues;
	fc_serialized->maxvalues = fc_range->target_maxvalues;

	/*
	 * 现在还要复制边界值（像长度计算一样，这取决于特定的数据类型）。
	 */
	fc_ptr = fc_serialized->data;		/* 序列化数据的开始 */

	for (fc_i = 0; fc_i < fc_nvalues; fc_i++)
	{
		if (fc_typbyval)			/* 简单的按值数据类型 */
		{
			Datum		fc_tmp;

			/*
			 * 对于按值类型，我们只需复制有效字节 -
			 * 我们不能直接使用 memcpy，因为它假设小端字节序
			 * 行为。store_att_byval 几乎能做我们需要的，但它
			 * 需要一个正确对齐的缓冲区 - 输出缓冲区并不
			 * 保证这一点。因此，我们仅使用一个本地 Datum 变量（这
			 * 可以保证正确对齐），然后从中复制值。
			 */
			store_att_byval(&fc_tmp, fc_range->values[fc_i], fc_typlen);

			memcpy(fc_ptr, &fc_tmp, fc_typlen);
			fc_ptr += fc_typlen;
		}
		else if (fc_typlen > 0)	/* 固定长度按引用类型 */
		{
			memcpy(fc_ptr, DatumGetPointer(fc_range->values[fc_i]), fc_typlen);
			fc_ptr += fc_typlen;
		}
		else if (fc_typlen == -1)	/* varlena */
		{
			int			fc_tmp = VARSIZE_ANY(DatumGetPointer(fc_range->values[fc_i]));

			memcpy(fc_ptr, DatumGetPointer(fc_range->values[fc_i]), fc_tmp);
			fc_ptr += fc_tmp;
		}
		else if (fc_typlen == -2)	/* cstring */
		{
			int			fc_tmp = strlen(DatumGetCString(fc_range->values[fc_i])) + 1;

			memcpy(fc_ptr, DatumGetCString(fc_range->values[fc_i]), fc_tmp);
			fc_ptr += fc_tmp;
		}

		/* 确保我们没有溢出缓冲区末尾 */
		Assert(fc_ptr <= ((char *) fc_serialized + fc_len));
	}

	/* 精确大小 */
	Assert(fc_ptr == ((char *) fc_serialized + fc_len));

	return fc_serialized;
}

/*
 * brin_range_deserialize
 *	  将内存中的表示序列化为紧凑的 varlena 值。
 *
 * 只需复制头部，然后复制存储在
 * 内存值数组中的各个值。
 */
static Ranges *
fc_brin_range_deserialize(int fc_maxvalues, SerializedRanges *fc_serialized)
{
	int			fc_i,
				fc_nvalues;
	char	   *fc_ptr,
			   *fc_dataptr;
	bool		fc_typbyval;
	int			fc_typlen;
	Size		fc_datalen;

	Ranges	   *fc_range;

	Assert(fc_serialized->nranges >= 0);
	Assert(fc_serialized->nvalues >= 0);
	Assert(fc_serialized->maxvalues > 0);

	fc_nvalues = 2 * fc_serialized->nranges + fc_serialized->nvalues;

	Assert(fc_nvalues <= fc_serialized->maxvalues);
	Assert(fc_serialized->maxvalues <= fc_maxvalues);

	fc_range = fc_minmax_multi_init(fc_maxvalues);

	/* 复制头部信息 */
	fc_range->nranges = fc_serialized->nranges;
	fc_range->nvalues = fc_serialized->nvalues;
	fc_range->nsorted = fc_serialized->nvalues;
	fc_range->maxvalues = fc_maxvalues;
	fc_range->target_maxvalues = fc_serialized->maxvalues;

	fc_range->typid = fc_serialized->typid;

	fc_typbyval = get_typbyval(fc_serialized->typid);
	fc_typlen = get_typlen(fc_serialized->typid);

	/*
	 * 现在将值拆分到 Datum 数组中。我们必须复制
	 * 数据，因为序列化表示会忽略对齐，我们
	 * 也不想依赖它被保留。
	 */
	fc_ptr = fc_serialized->data;

	/*
	 * 我们不想分配很多块，所以我们只在一块中分配所有内容。
	 * 我们需要多少空间？
	 *
	 * XXX 我们不需要复制简单的按值数据类型。
	 */
	fc_datalen = 0;
	fc_dataptr = NULL;
	for (fc_i = 0; (fc_i < fc_nvalues) && (!fc_typbyval); fc_i++)
	{
		if (fc_typlen > 0)			/* 固定长度按引用类型 */
			fc_datalen += MAXALIGN(fc_typlen);
		else if (fc_typlen == -1)	/* varlena */
		{
			fc_datalen += MAXALIGN(VARSIZE_ANY(DatumGetPointer(fc_ptr)));
			fc_ptr += VARSIZE_ANY(DatumGetPointer(fc_ptr));
		}
		else if (fc_typlen == -2)	/* cstring */
		{
			Size		fc_slen = strlen(DatumGetCString(fc_ptr)) + 1;

			fc_datalen += MAXALIGN(fc_slen);
			fc_ptr += fc_slen;
		}
	}

	if (fc_datalen > 0)
		fc_dataptr = palloc(fc_datalen);

	/*
	 * 恢复源指针（在计算我们需要分配的
	 * 空间时可能已被修改）。
	 */
	fc_ptr = fc_serialized->data;

	for (fc_i = 0; fc_i < fc_nvalues; fc_i++)
	{
		if (fc_typbyval)			/* 简单的按值数据类型 */
		{
			Datum		fc_v = 0;

			memcpy(&fc_v, fc_ptr, fc_typlen);

			fc_range->values[fc_i] = fetch_att(&fc_v, true, fc_typlen);
			fc_ptr += fc_typlen;
		}
		else if (fc_typlen > 0)	/* 固定长度按引用类型 */
		{
			fc_range->values[fc_i] = PointerGetDatum(fc_dataptr);

			memcpy(fc_dataptr, fc_ptr, fc_typlen);
			fc_dataptr += MAXALIGN(fc_typlen);

			fc_ptr += fc_typlen;
		}
		else if (fc_typlen == -1)	/* varlena */
		{
			fc_range->values[fc_i] = PointerGetDatum(fc_dataptr);

			memcpy(fc_dataptr, fc_ptr, VARSIZE_ANY(fc_ptr));
			fc_dataptr += MAXALIGN(VARSIZE_ANY(fc_ptr));
			fc_ptr += VARSIZE_ANY(fc_ptr);
		}
		else if (fc_typlen == -2)	/* cstring */
		{
			Size		fc_slen = strlen(fc_ptr) + 1;

			fc_range->values[fc_i] = PointerGetDatum(fc_dataptr);

			memcpy(fc_dataptr, fc_ptr, fc_slen);
			fc_dataptr += MAXALIGN(fc_slen);
			fc_ptr += fc_slen;
		}

		/* 确保我们没有溢出缓冲区末尾 */
		Assert(fc_ptr <= ((char *) fc_serialized + VARSIZE_ANY(fc_serialized)));
	}

	/* 应该准确消耗了整个输入值 */
	Assert(fc_ptr == ((char *) fc_serialized + VARSIZE_ANY(fc_serialized)));

	/* 返回反序列化值 */
	return fc_range;
}

/*
 * compare_expanded_ranges
 *	  比较扩展范围 - 首先按最小值，然后按最大值。
 *
 * 我们保证一个 Ranges 对象中的范围不重叠，
 * 所以仅按最小值排序似乎可能有些奇怪。但在合并两个
 * Ranges（在联合函数中发生）时，范围实际上可能重叠。
 * 所以我们会比较两者。
 */
static int fc_compare_expanded_ranges(const void *fc_a, const void *fc_b, void *fc_arg)
{
	ExpandedRange *fc_ra = (ExpandedRange *) fc_a;
	ExpandedRange *fc_rb = (ExpandedRange *) fc_b;
	Datum		fc_r;

	compare_context *fc_cxt = (compare_context *) fc_arg;

	/* 首先比较最小值 */
	fc_r = FunctionCall2Coll(fc_cxt->cmpFn, fc_cxt->colloid, fc_ra->minval, fc_rb->minval);

	if (DatumGetBool(fc_r))
		return -1;

	fc_r = FunctionCall2Coll(fc_cxt->cmpFn, fc_cxt->colloid, fc_rb->minval, fc_ra->minval);

	if (DatumGetBool(fc_r))
		return 1;

	/* 然后比较最大值 */
	fc_r = FunctionCall2Coll(fc_cxt->cmpFn, fc_cxt->colloid, fc_ra->maxval, fc_rb->maxval);

	if (DatumGetBool(fc_r))
		return -1;

	fc_r = FunctionCall2Coll(fc_cxt->cmpFn, fc_cxt->colloid, fc_rb->maxval, fc_ra->maxval);

	if (DatumGetBool(fc_r))
		return 1;

	return 0;
}

/*
 * compare_values
 *	  比较值。
 */
static int fc_compare_values(const void *fc_a, const void *fc_b, void *fc_arg)
{
	Datum	   *fc_da = (Datum *) fc_a;
	Datum	   *fc_db = (Datum *) fc_b;
	Datum		fc_r;

	compare_context *fc_cxt = (compare_context *) fc_arg;

	fc_r = FunctionCall2Coll(fc_cxt->cmpFn, fc_cxt->colloid, *fc_da, *fc_db);

	if (DatumGetBool(fc_r))
		return -1;

	fc_r = FunctionCall2Coll(fc_cxt->cmpFn, fc_cxt->colloid, *fc_db, *fc_da);

	if (DatumGetBool(fc_r))
		return 1;

	return 0;
}

/*
 * 检查新值是否匹配现有范围中的一个。
 */
static bool fc_has_matching_range(BrinDesc *fc_bdesc, Oid fc_colloid, Ranges *fc_ranges,
				   Datum fc_newval, AttrNumber fc_attno, Oid fc_typid)
{
	Datum		fc_compar;

	Datum		fc_minvalue;
	Datum		fc_maxvalue;

	FmgrInfo   *fc_cmpLessFn;
	FmgrInfo   *fc_cmpGreaterFn;

	/* 在范围上进行二分搜索 */
	int			fc_start,
				fc_end;

	if (fc_ranges->nranges == 0)
		return false;

	fc_minvalue = fc_ranges->values[0];
	fc_maxvalue = fc_ranges->values[2 * fc_ranges->nranges - 1];

	/*
	 * 否则，需要将新值与所有范围的边界进行比较。首先检查它是否小于绝对最小值，即数组中的第一个值。
	 */
	fc_cmpLessFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_typid,
												   BTLessStrategyNumber);
	fc_compar = FunctionCall2Coll(fc_cmpLessFn, fc_colloid, fc_newval, fc_minvalue);

	/* 小于范围列表中的最小值 */
	if (DatumGetBool(fc_compar))
		return false;

	/*
	 * 现在将其与现有最大值（数据数组中的最后一个值）进行比较。但前提是我们尚未在 minvalue 检查中排除可能的匹配。
	 */
	fc_cmpGreaterFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_typid,
													  BTGreaterStrategyNumber);
	fc_compar = FunctionCall2Coll(fc_cmpGreaterFn, fc_colloid, fc_newval, fc_maxvalue);

	if (DatumGetBool(fc_compar))
		return false;

	/*
	 * 所以我们知道它在一般的最小/最大范围内，问题是它是否落在某个范围或间隙中。我们将对各个范围进行二分搜索 - 对于每个范围，我们检查相等（值是否落入该范围），然后检查当前范围上方或下方的范围。
	 */
	fc_start = 0;					/* 第一个范围 */
	fc_end = (fc_ranges->nranges - 1);	/* 最后一个范围 */
	while (true)
	{
		int			fc_midpoint = (fc_start + fc_end) / 2;

		/* 这意味着我们在上一步中耗尽了范围 */
		if (fc_start > fc_end)
			return false;

		/* 复制范围的最小/最大值 */
		fc_minvalue = fc_ranges->values[2 * fc_midpoint];
		fc_maxvalue = fc_ranges->values[2 * fc_midpoint + 1];

		/*
		 * 该值是否小于 minval？如果是，我们将递归到范围数组的左侧。
		 */
		fc_compar = FunctionCall2Coll(fc_cmpLessFn, fc_colloid, fc_newval, fc_minvalue);

		/* 小于此范围内的最小值 */
		if (DatumGetBool(fc_compar))
		{
			fc_end = (fc_midpoint - 1);
			continue;
		}

		/*
		 * 该值是否大于 minval？如果是，我们将递归到范围数组的右侧。
		 */
		fc_compar = FunctionCall2Coll(fc_cmpGreaterFn, fc_colloid, fc_newval, fc_maxvalue);

		/* 大于此范围内的最大值 */
		if (DatumGetBool(fc_compar))
		{
			fc_start = (fc_midpoint + 1);
			continue;
		}

		/* 嘿，我们找到了匹配的范围 */
		return true;
	}

	return false;
}


/*
 * range_contains_value
 * 		检查新值是否已经包含在范围列表中。
 *
 * 我们首先检查区间列表。我们使用一个小技巧 - 我们首先将该值与整个范围的最小值/最大值（第一个区间的最小值，最后一个区间的最大值）进行比较，只有在通过后才检查单个区间。
 *
 * 如果该值与任何区间都不匹配，我们检查确切值。我们简单地循环遍历它们，并对它们调用相等运算符。
 *
 * 最后一个参数（full）决定是否需要搜索所有值，包括未排序的部分。若 full=false，则不会搜索未排序的部分，这可能会产生假阴性和重复值（仅在未排序部分中），但在构建范围时这样没有问题 - 我们将在序列化之前去重，这只能发生在存在未排序值时（即它已被修改）。
 *
 * 序列化的范围不包含任何未排序的值，因此在查询期间无法导致假阴性。
 */
static bool fc_range_contains_value(BrinDesc *fc_bdesc, Oid fc_colloid,
					 AttrNumber fc_attno, Form_pg_attribute fc_attr,
					 Ranges *fc_ranges, Datum fc_newval, bool fc_full)
{
	int			fc_i;
	FmgrInfo   *fc_cmpEqualFn;
	Oid			fc_typid = fc_attr->atttypid;

	/*
	 * 首先检查范围，如果有的话。我们先检查整个范围，只有在仍有机会找到匹配时才检查各个范围。
	 */
	if (fc_has_matching_range(fc_bdesc, fc_colloid, fc_ranges, fc_newval, fc_attno, fc_typid))
		return true;

	fc_cmpEqualFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_typid,
													BTEqualStrategyNumber);

	/*
	 * 没有匹配范围，所以让我们检查排序后的值。
	 *
	 * 我们对少量值进行顺序搜索，对于超过16个值的情况则进行二分搜索。
	 * 这个阈值有些任意，因为它取决于比较函数的开销。
	 *
	 * XXX 如果我们在这里使用阈值，也许应该在has_matching_range中做同样的事情？
	 * 或者也许我们应该始终进行二分搜索？
	 *
	 * XXX 我们可以使用与范围的相同优化，检查值是否在min/max之间，
	 * 这样也许可以排除所有排序的值，而不必检查所有这些值。
	 */
	if (fc_ranges->nsorted >= 16)
	{
		compare_context fc_cxt;

		fc_cxt.colloid = fc_ranges->colloid;
		fc_cxt.cmpFn = fc_ranges->cmp;

		if (bsearch_arg(&fc_newval, &fc_ranges->values[2 * fc_ranges->nranges],
						fc_ranges->nsorted, sizeof(Datum),
						fc_compare_values, (void *) &fc_cxt) != NULL)
			return true;
	}
	else
	{
		for (fc_i = 2 * fc_ranges->nranges; fc_i < 2 * fc_ranges->nranges + fc_ranges->nsorted; fc_i++)
		{
			Datum		fc_compar;

			fc_compar = FunctionCall2Coll(fc_cmpEqualFn, fc_colloid, fc_newval, fc_ranges->values[fc_i]);

			/* 找到一个确切的匹配 */
			if (DatumGetBool(fc_compar))
				return true;
		}
	}

	/* 如果没有要求检查未排序部分，我们就完成了。 */
	if (!fc_full)
		return false;

	/* 检查未排序部分。 */
	for (fc_i = 2 * fc_ranges->nranges + fc_ranges->nsorted; fc_i < 2 * fc_ranges->nranges + fc_ranges->nvalues; fc_i++)
	{
		Datum		fc_compar;

		fc_compar = FunctionCall2Coll(fc_cmpEqualFn, fc_colloid, fc_newval, fc_ranges->values[fc_i]);

		/* 找到一个确切的匹配 */
		if (DatumGetBool(fc_compar))
			return true;
	}

	/* 该值不被此BRIN元组覆盖 */
	return false;
}

/*
 * 从Ranges扩展范围到ExpandedRange数组。这要求eranges预先分配并具有正确的大小 -
 * 需要(nranges + nvalues)个元素。
 *
 * 扩展范围的顺序是任意的。我们首先扩展范围，这部分是排序的。
 * 然后我们扩展值，这部分可能是未排序的。
 */
static void fc_fill_expanded_ranges(ExpandedRange *fc_eranges, int fc_neranges, Ranges *fc_ranges)
{
	int			fc_idx;
	int			fc_i;

	/* 检查输出数组是否具有正确的大小。 */
	Assert(fc_neranges == (fc_ranges->nranges + fc_ranges->nvalues));

	fc_idx = 0;
	for (fc_i = 0; fc_i < fc_ranges->nranges; fc_i++)
	{
		fc_eranges[fc_idx].minval = fc_ranges->values[2 * fc_i];
		fc_eranges[fc_idx].maxval = fc_ranges->values[2 * fc_i + 1];
		fc_eranges[fc_idx].collapsed = false;
		fc_idx++;

		Assert(fc_idx <= fc_neranges);
	}

	for (fc_i = 0; fc_i < fc_ranges->nvalues; fc_i++)
	{
		fc_eranges[fc_idx].minval = fc_ranges->values[2 * fc_ranges->nranges + fc_i];
		fc_eranges[fc_idx].maxval = fc_ranges->values[2 * fc_ranges->nranges + fc_i];
		fc_eranges[fc_idx].collapsed = true;
		fc_idx++;

		Assert(fc_idx <= fc_neranges);
	}

	/* 我们是否产生了预期数量的元素？ */
	Assert(fc_idx == fc_neranges);

	return;
}

/*
 * 对扩展范围进行排序和去重。
 *
 * 范围可能会被去重 - 我们只是简单地附加值，而不检查重复等。因此，去重可能会使范围数量减少到足够，
 * 我们将不必计算距离等。
 *
 * 返回扩展范围的数量。
 */
static int fc_sort_expanded_ranges(FmgrInfo *fc_cmp, Oid fc_colloid,
					 ExpandedRange *fc_eranges, int fc_neranges)
{
	int			fc_n;
	int			fc_i;
	compare_context fc_cxt;

	Assert(fc_neranges > 0);

	/* 对值进行排序 */
	fc_cxt.colloid = fc_colloid;
	fc_cxt.cmpFn = fc_cmp;

	/*
	 * XXX 我们对所有值执行qsort，但我们也可以利用某些输入数据已经排序的事实
	 * （所有范围，可能还有一些点）并执行归并排序。
	 */
	qsort_arg(fc_eranges, fc_neranges, sizeof(ExpandedRange),
			  fc_compare_expanded_ranges, (void *) &fc_cxt);

	/*
	 * 对范围进行去重 - 只需将每个范围与前一个进行比较，跳过重复的范围。
	 */
	fc_n = 1;
	for (fc_i = 1; fc_i < fc_neranges; fc_i++)
	{
		/* 如果当前范围等于前一个范围，则不做任何操作 */
		if (!fc_compare_expanded_ranges(&fc_eranges[fc_i - 1], &fc_eranges[fc_i], (void *) &fc_cxt))
			continue;

		/* 否则，将其复制到第n个位置（如果尚未在那里） */
		if (fc_i != fc_n)
			memcpy(&fc_eranges[fc_n], &fc_eranges[fc_i], sizeof(ExpandedRange));

		fc_n++;
	}

	Assert((fc_n > 0) && (fc_n <= fc_neranges));

	return fc_n;
}

/*
 * 在组合多个Range值（在union函数中）时，一些范围可能重叠。
 * 我们简单地合并重叠的范围来解决这个问题。
 *
 * XXX 这假设扩展的范围以前是按minval和maxval排序的。
 * 当检测重叠时，我们利用这一点。
 */
static int fc_merge_overlapping_ranges(FmgrInfo *fc_cmp, Oid fc_colloid,
						 ExpandedRange *fc_eranges, int fc_neranges)
{
	int			fc_idx;

	/* 如果范围(idx)和(idx+1)重叠，则合并它们。 */
	fc_idx = 0;
	while (fc_idx < (fc_neranges - 1))
	{
		Datum		fc_r;

		/*
		 * 比较[?,maxval]与[minval,?] - 如果(minval < maxval)，范围重叠
		 */
		fc_r = FunctionCall2Coll(fc_cmp, fc_colloid,
							  fc_eranges[fc_idx].maxval,
							  fc_eranges[fc_idx + 1].minval);

		/*
		 * 不，maxval < minval，所以没有重叠。而且我们知道范围是排好序的，
		 * 所以没有更多的重叠，因为所有剩余的范围都有更大或相等的minval。
		 */
		if (DatumGetBool(fc_r))
		{
			/* 继续到下一个范围 */
			fc_idx += 1;
			continue;
		}

		/*
		 * 因此范围 'idx' 和 'idx+1' 确实重叠，但我们不知道
		 * 'idx+1' 是否包含在 'idx' 中，或者它们是否仅部分重叠。
		 * 因此比较上界并保留较大的一个。
		 */
		fc_r = FunctionCall2Coll(fc_cmp, fc_colloid,
							  fc_eranges[fc_idx].maxval,
							  fc_eranges[fc_idx + 1].maxval);

		if (DatumGetBool(fc_r))
			fc_eranges[fc_idx].maxval = fc_eranges[fc_idx + 1].maxval;

		/*
		 * 该范围肯定不再是重叠的（无论之前的状态如何）。
		 */
		fc_eranges[fc_idx].collapsed = false;

		/*
		 * 现在通过将剩余范围全部向前移动 1 个位置来完全去除 (idx+1) 范围。
		 * 有 neranges 个元素，我们需要移动来自 (idx+2) 的元素。
		 * 这意味着要移动的元素数量为 [ncranges - (idx+2)]。
		 */
		memmove(&fc_eranges[fc_idx + 1], &fc_eranges[fc_idx + 2],
				(fc_neranges - (fc_idx + 2)) * sizeof(ExpandedRange));

		/*
		 * 减少范围的数量，并重复（使用相同的范围，因为由于合并
		 * 它可能与额外的范围重叠）。
		 */
		fc_neranges--;
	}

	return fc_neranges;
}

/*
 * 距离值的简单比较器，比较双精度值。
 * 这故意按降序排列距离，即较长的间隙将位于前面。
 */
static int fc_compare_distances(const void *fc_a, const void *fc_b)
{
	DistanceValue *fc_da = (DistanceValue *) fc_a;
	DistanceValue *fc_db = (DistanceValue *) fc_b;

	if (fc_da->value < fc_db->value)
		return 1;
	else if (fc_da->value > fc_db->value)
		return -1;

	return 0;
}

/*
 * 给定一个扩展范围的数组，计算每个范围之间的间隙大小。
 * 对于 neranges，有 (neranges-1) 个间隙。
 *
 * 我们简单地调用“距离”函数来计算连续范围对的 (max-min)。
 * 该函数可能比较昂贵，因此我们只执行一次（然后将其用于尽可能
 * 多地选择合并的范围）。
 *
 * 有关详细信息，请参见 reduce_expanded_ranges。
 */
static DistanceValue *
fc_build_distances(FmgrInfo *fc_distanceFn, Oid fc_colloid,
				ExpandedRange *fc_eranges, int fc_neranges)
{
	int			fc_i;
	int			fc_ndistances;
	DistanceValue *fc_distances;

	Assert(fc_neranges > 0);

	/* 如果只有一个范围，则没有距离可计算。 */
	if (fc_neranges == 1)
		return NULL;

	fc_ndistances = (fc_neranges - 1);
	fc_distances = (DistanceValue *) palloc0(sizeof(DistanceValue) * fc_ndistances);

	/*
	 * 遍历范围并计算范围之间的距离，以便我们可以将它们排序一次。
	 */
	for (fc_i = 0; fc_i < fc_ndistances; fc_i++)
	{
		Datum		fc_a1,
					fc_a2,
					fc_r;

		fc_a1 = fc_eranges[fc_i].maxval;
		fc_a2 = fc_eranges[fc_i + 1].minval;

		/* 计算间隙的长度（在 max/min 之间） */
		fc_r = FunctionCall2Coll(fc_distanceFn, fc_colloid, fc_a1, fc_a2);

		/* 记住距离所对应的间隙索引 */
		fc_distances[fc_i].index = fc_i;
		fc_distances[fc_i].value = DatumGetFloat8(fc_r);
	}

	/*
	 * 按降序排序距离，这样最长的间隙位于前面。
	 */
	pg_qsort(fc_distances, fc_ndistances, sizeof(DistanceValue), fc_compare_distances);

	return fc_distances;
}

/*
 * 为现有范围（和单点范围）构建扩展范围，以及不适合
 * 数组的新值。这种扩展表示法使处理变得更容易，因为它允许
 * 以相同的方式处理范围和点。
 *
 * 我们对扩展范围进行排序和去重 - 这是必要的，因为
 * 点可能是无序的。此外这两个部分（范围和
 * 点）各自排序。
 */
static ExpandedRange *
fc_build_expanded_ranges(FmgrInfo *fc_cmp, Oid fc_colloid, Ranges *fc_ranges,
					  int *fc_nranges)
{
	int			fc_neranges;
	ExpandedRange *fc_eranges;

	/* 范围和点都扩展为单独的元素 */
	fc_neranges = fc_ranges->nranges + fc_ranges->nvalues;

	fc_eranges = (ExpandedRange *) palloc0(fc_neranges * sizeof(ExpandedRange));

	/* 填充扩展范围 */
	fc_fill_expanded_ranges(fc_eranges, fc_neranges, fc_ranges);

	/* 排序并去重扩展范围 */
	fc_neranges = fc_sort_expanded_ranges(fc_cmp, fc_colloid, fc_eranges, fc_neranges);

	/* 记住我们构建了多少个范围 */
	*fc_nranges = fc_neranges;

	return fc_eranges;
}

#ifdef USE_ASSERT_CHECKING
/*
 * 计算存储范围所需的边界值。每个单点范围
 * 使用一个值存储，每个常规范围需要两个值。
 */
static int fc_count_values(ExpandedRange *fc_cranges, int fc_ncranges)
{
	int			fc_i;
	int			fc_count;

	fc_count = 0;
	for (fc_i = 0; fc_i < fc_ncranges; fc_i++)
	{
		if (fc_cranges[fc_i].collapsed)
			fc_count += 1;
		else
			fc_count += 2;
	}

	return fc_count;
}
#endif

/*
 * reduce_expanded_ranges
 *		减少范围，直到值的数量足够低
 *
 * 组合范围，直到边界值的数量降到由 max_values 指定的阈值以下。通过根据它们之间的距离合并足够的范围来实现。
 *
 * 返回结果范围的数量。
 *
 * 我们简单地使用全局的最小/最大值，然后为足够大的间隙添加边界。每个间隙增加 2 个值，因此我们简单地使用 (target/2-1) 的距离。然后我们简单地对所有值进行排序 - 每两个值是一个范围的边界（可能被折叠）。
 *
 * XXX 一些范围可能被折叠（即最小/最大值相等），但我们暂时忽略这一点。我们可以重复这个过程，递归地添加几个间隙。
 *
 * XXX 合并的范围完全是根据距离选择的。但这可能不是最佳策略，例如当多个间隙的长度相等（或非常相似）时。
 *
 * 例如考虑点 1，2，3，..，64，它们当然有相同长度为 1 的间隙。在这种情况下，我们倾向于选择第一段该长度的间隙，这导致了这样的结果：
 *
 *    第一步：  [1, 2], 3, 4, 5, .., 64
 *    第二步：  [1, 3], 4, 5,    .., 64
 *    第三步：  [1, 4], 5,       .., 64
 *    ...
 *
 * 所以最后我们将会有一个“大的”范围和多个小点。这可能可以，但看起来有点奇怪且不够优化。也许在选择要合并的范围时我们应该考虑其他因素 - 例如范围的长度？或者或许随机选择范围，选择的概率与距离成反比（间隙长度可能非常接近，但不完全相同）。
 *
 * XXX 或者也许我们可以通过使用随机值作为平局打破的方法来处理这个问题，或通过向实际距离添加随机噪声。
 */
static int fc_reduce_expanded_ranges(ExpandedRange *fc_eranges, int fc_neranges,
					   DistanceValue *fc_distances, int fc_max_values,
					   FmgrInfo *fc_cmp, Oid fc_colloid)
{
	int			fc_i;
	int			fc_nvalues;
	Datum	   *fc_values;

	compare_context fc_cxt;

	/* 范围之间的总间隙数 */
	int			fc_ndistances = (fc_neranges - 1);

	/* 要保留的间隙数 */
	int			fc_keep = (fc_max_values / 2 - 1);

	/*
	 * 也许我们已经有足够少的范围了？
	 *
	 * XXX 这应该在我们真正做诸如排序等耗费时间的操作之前进行，所以这也许应该仅仅是一个断言。
	 */
	if (fc_keep >= fc_ndistances)
		return fc_neranges;

	/* 对值进行排序 */
	fc_cxt.colloid = fc_colloid;
	fc_cxt.cmpFn = fc_cmp;

	/* 为边界值分配空间 */
	fc_nvalues = 0;
	fc_values = (Datum *) palloc(sizeof(Datum) * fc_max_values);

	/* 添加来自第一个/最后一个范围的全局最小/最大值 */
	fc_values[fc_nvalues++] = fc_eranges[0].minval;
	fc_values[fc_nvalues++] = fc_eranges[fc_neranges - 1].maxval;

	/* 为足够的间隙添加边界值 */
	for (fc_i = 0; fc_i < fc_keep; fc_i++)
	{
		/* (index) 和 (index+1) 范围之间的间隙索引 */
		int			fc_index = fc_distances[fc_i].index;

		Assert((fc_index >= 0) && ((fc_index + 1) < fc_neranges));

		/* 添加前一个范围的最大值，后一个范围的最小值 */
		fc_values[fc_nvalues++] = fc_eranges[fc_index].maxval;
		fc_values[fc_nvalues++] = fc_eranges[fc_index + 1].minval;

		Assert(fc_nvalues <= fc_max_values);
	}

	/* 我们应该有一个偶数数量的范围值。 */
	Assert(fc_nvalues % 2 == 0);

	/*
	 * 使用比较函数对值进行排序，并从排序结果中形成范围。
	 */
	qsort_arg(fc_values, fc_nvalues, sizeof(Datum),
			  fc_compare_values, (void *) &fc_cxt);

	/* 我们有 nvalues 边界值，这意味着 nvalues/2 范围。 */
	for (fc_i = 0; fc_i < (fc_nvalues / 2); fc_i++)
	{
		fc_eranges[fc_i].minval = fc_values[2 * fc_i];
		fc_eranges[fc_i].maxval = fc_values[2 * fc_i + 1];

		/* 如果边界值相同，则这就是一个被折叠的范围 */
		fc_eranges[fc_i].collapsed = (fc_compare_values(&fc_values[2 * fc_i],
											   &fc_values[2 * fc_i + 1],
											   &fc_cxt) == 0);
	}

	return (fc_nvalues / 2);
}

/*
 * 将 ExpandedRanges 中的边界值存储回 'ranges'（仅使用所需的最小值）。
 */
static void fc_store_expanded_ranges(Ranges *fc_ranges, ExpandedRange *fc_eranges, int fc_neranges)
{
	int			fc_i;
	int			fc_idx = 0;

	/* 首先复制常规范围 */
	fc_ranges->nranges = 0;
	for (fc_i = 0; fc_i < fc_neranges; fc_i++)
	{
		if (!fc_eranges[fc_i].collapsed)
		{
			fc_ranges->values[fc_idx++] = fc_eranges[fc_i].minval;
			fc_ranges->values[fc_idx++] = fc_eranges[fc_i].maxval;
			fc_ranges->nranges++;
		}
	}

	/* 现在复制被折叠的范围 */
	fc_ranges->nvalues = 0;
	for (fc_i = 0; fc_i < fc_neranges; fc_i++)
	{
		if (fc_eranges[fc_i].collapsed)
		{
			fc_ranges->values[fc_idx++] = fc_eranges[fc_i].minval;
			fc_ranges->nvalues++;
		}
	}

	/* 所有值都是排序的 */
	fc_ranges->nsorted = fc_ranges->nvalues;

	Assert(fc_count_values(fc_eranges, fc_neranges) == 2 * fc_ranges->nranges + fc_ranges->nvalues);
	Assert(2 * fc_ranges->nranges + fc_ranges->nvalues <= fc_ranges->maxvalues);
}


/*
 * 考虑在范围内释放空间。检查是否有足够的空间容纳至少一个新值，并在必要时进行压缩。
 *
 * 如果值实际上被修改，则返回true。
 */
static bool fc_ensure_free_space_in_buffer(BrinDesc *fc_bdesc, Oid fc_colloid,
							AttrNumber fc_attno, Form_pg_attribute fc_attr,
							Ranges *fc_range)
{
	MemoryContext fc_ctx;
	MemoryContext fc_oldctx;

	FmgrInfo   *fc_cmpFn,
			   *fc_distanceFn;

	/* 扩展范围 */
	ExpandedRange *fc_eranges;
	int			fc_neranges;
	DistanceValue *fc_distances;

	/*
	 * 如果缓冲区中有空闲空间，我们就可以完成，而无需修改任何内容。
	 */
	if (2 * fc_range->nranges + fc_range->nvalues < fc_range->maxvalues)
		return false;

	/* 我们肯定需要比较器，所以现在就查找它 */
	fc_cmpFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
											   BTLessStrategyNumber);

	/* 如果有未排序的部分，则去重值 */
	fc_range_deduplicate_values(fc_range);

	/*
	 * 通过去重，我们是否减少了足够的空闲空间？
	 *
	 * 我们不简单地再次检查range->maxvalues。去重可能释放的空间非常有限（例如，仅一个值），这可能迫使我们非常频繁地去重。在这种情况下，进行压缩以释放更多空间会更好。
	 */
	if (2 * fc_range->nranges + fc_range->nvalues <= fc_range->maxvalues * MINMAX_BUFFER_LOAD_FACTOR)
		return true;

	/*
	 * 我们需要合并一些现有的范围，以减少必须存储的值的数量。
	 *
	 * distanceFn调用（可能内部调用例如numeric_le）可能会分配相当多的内存，我们必须不泄漏内存（即使对于单个BRIN页面范围，我们可能不得不重复进行此操作）。否则，当构建新索引时，可能会出现问题。因此，我们使用内存上下文，并确保在结束时释放内存（如果我们多次调用距离函数，可能会出现问题，但算了）。
	 */
	fc_ctx = AllocSetContextCreate(CurrentMemoryContext,
								"minmax-multi context",
								ALLOCSET_DEFAULT_SIZES);

	fc_oldctx = MemoryContextSwitchTo(fc_ctx);

	/* 构建扩展范围 */
	fc_eranges = fc_build_expanded_ranges(fc_cmpFn, fc_colloid, fc_range, &fc_neranges);

	/* 我们还需要“距离”过程 */
	fc_distanceFn = fc_minmax_multi_get_procinfo(fc_bdesc, fc_attno, PROCNUM_DISTANCE);

	/* 构建间隙距离数组并按升序排序 */
	fc_distances = fc_build_distances(fc_distanceFn, fc_colloid, fc_eranges, fc_neranges);

	/*
	 * 合并范围，直到我们释放至少50%的空间。这个阈值有些任意，可能需要调整。我们不能使用过低或过高的值。
	 */
	fc_neranges = fc_reduce_expanded_ranges(fc_eranges, fc_neranges, fc_distances,
									  fc_range->maxvalues * MINMAX_BUFFER_LOAD_FACTOR,
									  fc_cmpFn, fc_colloid);

	/* 确保我们已充分减少范围的数量。 */
	Assert(fc_count_values(fc_eranges, fc_neranges) <= fc_range->maxvalues * MINMAX_BUFFER_LOAD_FACTOR);

	/* 将扩展范围分解为常规范围和单个值 */
	fc_store_expanded_ranges(fc_range, fc_eranges, fc_neranges);

	MemoryContextSwitchTo(fc_oldctx);
	MemoryContextDelete(fc_ctx);

	/* 我们是否以某种方式打破了范围？ */
	fc_AssertCheckRanges(fc_range, fc_cmpFn, fc_colloid);

	return true;
}

/*
 * range_add_value
 * 		将新值添加到minmax-multi范围中。
 */
static bool fc_range_add_value(BrinDesc *fc_bdesc, Oid fc_colloid,
				AttrNumber fc_attno, Form_pg_attribute fc_attr,
				Ranges *fc_ranges, Datum fc_newval)
{
	FmgrInfo   *fc_cmpFn;
	bool		fc_modified = false;

	/* 我们肯定需要比较器，所以现在就查找它 */
	fc_cmpFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
											   BTLessStrategyNumber);

	/* 对输入范围进行全面检查 */
	fc_AssertCheckRanges(fc_ranges, fc_cmpFn, fc_colloid);

	/*
	 * 确保缓冲区中有足够的空闲空间。我们仅在缓冲区已满时触发此操作，这意味着它必须被修改，因为我们将其大小调整为大于存储在磁盘上的内容。
	 *
	 * 这需要在检查值是否包含在范围内之前进行，因为该值可能在未排序的部分，而我们并没有在range_contains_value中检查这一点。去重将其移动到已排序部分，我们也会添加该值，这违反了我们从不在范围或已排序值中有重复项的规则。
	 *
	 * 我们也可以去重并重新检查值是否包含，但这似乎过于繁琐。我们无论如何都需要去重，所以为什么不现在就做。
	 */
	fc_modified = fc_ensure_free_space_in_buffer(fc_bdesc, fc_colloid,
										   fc_attno, fc_attr, fc_ranges);

	/*
	 * 如果值已经被范围覆盖，则退出。
	 *
	 * 我们也可以添加值直到达到 values_per_range，然后进行
	 * 批量去重，希望能获得更好的效率。但这意味着我们每次
	 * 实际上都修改了范围，这意味着必须对值进行序列化，
	 * 这需要 palloc、遍历值、复制它们等操作。并不是很便宜。
	 *
	 * 所以我们检查一下，该操作应该相当便宜——假设比较
	 * 函数不是很昂贵。
	 *
	 * 这也意味着值数组不能包含重复值。
	 */
	if (fc_range_contains_value(fc_bdesc, fc_colloid, fc_attno, fc_attr, fc_ranges, fc_newval, false))
		return fc_modified;

	/* 如果需要，复制值。 */
	fc_newval = datumCopy(fc_newval, fc_attr->attbyval, fc_attr->attlen);

	/*
	 * 如果值数组有空间，就复制进去，然后完成。
	 *
	 * 我们确实希望保持值的排序（加速搜索），所以我们
	 * 进行简单的插入排序。我们可以做一些更复杂的操作，
	 * 例如只偶尔对值进行排序，但对于小数量（例如当
	 * maxvalues 为 64 时）这应该就足够了。
	 */
	fc_ranges->values[2 * fc_ranges->nranges + fc_ranges->nvalues] = fc_newval;
	fc_ranges->nvalues++;

	/* 如果我们添加了第一个值，则可以认为它是排序的。 */
	if (fc_ranges->nvalues == 1)
		fc_ranges->nsorted = 1;

	/*
	 * 检查我们是否破坏了边界值的顺序（检查两个部分，
	 * 但这无妨）。
	 */
	fc_AssertCheckRanges(fc_ranges, fc_cmpFn, fc_colloid);

	/* 检查范围是否包含我们刚刚添加的值。 */
	Assert(fc_range_contains_value(fc_bdesc, fc_colloid, fc_attno, fc_attr, fc_ranges, fc_newval, true));

	/* 是的，我们已经修改了范围 */
	return true;
}

/*
 * 生成在“批处理模式”下收集的数据的范围表示。
 * 这类似于 reduce_expanded_ranges，除了我们不能假设
 * 值是排序的，并且可能存在重复值。
 */
static void fc_compactify_ranges(BrinDesc *fc_bdesc, Ranges *fc_ranges, int fc_max_values)
{
	FmgrInfo   *fc_cmpFn,
			   *fc_distanceFn;

	/* 扩展范围 */
	ExpandedRange *fc_eranges;
	int			fc_neranges;
	DistanceValue *fc_distances;

	MemoryContext fc_ctx;
	MemoryContext fc_oldctx;

	/*
	 * 我们需要实际压缩任何东西吗？
	 *
	 * 压缩可能需要的原因有两个——首先，值可能太多，
	 * 或者某些值可能是未排序的。
	 */
	if ((fc_ranges->nranges * 2 + fc_ranges->nvalues <= fc_max_values) &&
		(fc_ranges->nsorted == fc_ranges->nvalues))
		return;

	/* 我们肯定需要比较器，所以现在就查找它 */
	fc_cmpFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_ranges->attno, fc_ranges->typid,
											   BTLessStrategyNumber);

	/* 我们还需要“距离”过程 */
	fc_distanceFn = fc_minmax_multi_get_procinfo(fc_bdesc, fc_ranges->attno, PROCNUM_DISTANCE);

	/*
	 * distanceFn 调用（可能会内部调用 e.g. numeric_le）可能
	 * 会分配相当多的内存，我们必须避免泄漏。否则，
	 * 我们在构建索引时会遇到问题。因此我们创建一个
	 * 本地内存上下文，并确保在离开这个函数之前释放内存
	 * （而不是在每次调用后）。
	 */
	fc_ctx = AllocSetContextCreate(CurrentMemoryContext,
								"minmax-multi context",
								ALLOCSET_DEFAULT_SIZES);

	fc_oldctx = MemoryContextSwitchTo(fc_ctx);

	/* 构建扩展范围 */
	fc_eranges = fc_build_expanded_ranges(fc_cmpFn, fc_ranges->colloid, fc_ranges, &fc_neranges);

	/* 构建间隙距离数组并按升序排序 */
	fc_distances = fc_build_distances(fc_distanceFn, fc_ranges->colloid,
								fc_eranges, fc_neranges);

	/*
	 * 合并范围，直到我们低于 max_values。我们不使用任何
	 * 缩放因子，因为这在序列化期间使用，并且我们不期望
	 * 很快再插入更多元组。
	 */
	fc_neranges = fc_reduce_expanded_ranges(fc_eranges, fc_neranges, fc_distances,
									  fc_max_values, fc_cmpFn, fc_ranges->colloid);

	Assert(fc_count_values(fc_eranges, fc_neranges) <= fc_max_values);

	/* 转换回常规范围和单个值 */
	fc_store_expanded_ranges(fc_ranges, fc_eranges, fc_neranges);

	/* 检查所有范围不变量 */
	fc_AssertCheckRanges(fc_ranges, fc_cmpFn, fc_ranges->colloid);

	MemoryContextSwitchTo(fc_oldctx);
	MemoryContextDelete(fc_ctx);
}

Datum brin_minmax_multi_opcinfo(PG_FUNCTION_ARGS)
{
	BrinOpcInfo *fc_result;

	/*
	 * opaque->strategy_procinfos 是懒初始化的；在这里，通过 palloc0
	 * 将其设置为全未初始化，fn_oid 被设置为 InvalidOid。
	 */

	fc_result = palloc0(MAXALIGN(SizeofBrinOpcInfo(1)) +
					 sizeof(MinmaxMultiOpaque));
	fc_result->oi_nstored = 1;
	fc_result->oi_regular_nulls = true;
	fc_result->oi_opaque = (MinmaxMultiOpaque *)
		MAXALIGN((char *) fc_result + SizeofBrinOpcInfo(1));
	fc_result->oi_typcache[0] = lookup_type_cache(PG_BRIN_MINMAX_MULTI_SUMMARYOID, 0);

	PG_RETURN_POINTER(fc_result);
}

/*
 * 计算两个 float4 值之间的距离（普通减法）。
 */
Datum brin_minmax_multi_distance_float4(PG_FUNCTION_ARGS)
{
	float		fc_a1 = PG_GETARG_FLOAT4(0);
	float		fc_a2 = PG_GETARG_FLOAT4(1);

	/* 如果两个值都是 NaN，则我们将其视为相同 */
	if (isnan(fc_a1) && isnan(fc_a2))
		PG_RETURN_FLOAT8(0.0);

	/* 如果一个值是 NaN，则使用无限距离 */
	if (isnan(fc_a1) || isnan(fc_a2))
		PG_RETURN_FLOAT8(get_float8_infinity());

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(fc_a1 <= fc_a2);

	PG_RETURN_FLOAT8((double) fc_a2 - (double) fc_a1);
}

/*
 * 计算两个 float8 值之间的距离（普通减法）。
 */
Datum brin_minmax_multi_distance_float8(PG_FUNCTION_ARGS)
{
	double		fc_a1 = PG_GETARG_FLOAT8(0);
	double		fc_a2 = PG_GETARG_FLOAT8(1);

	/* 如果两个值都是 NaN，则我们将其视为相同 */
	if (isnan(fc_a1) && isnan(fc_a2))
		PG_RETURN_FLOAT8(0.0);

	/* 如果一个值是 NaN，则使用无限距离 */
	if (isnan(fc_a1) || isnan(fc_a2))
		PG_RETURN_FLOAT8(get_float8_infinity());

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(fc_a1 <= fc_a2);

	PG_RETURN_FLOAT8(fc_a2 - fc_a1);
}

/*
 * 计算两个 int2 值之间的距离（普通减法）。
 */
Datum brin_minmax_multi_distance_int2(PG_FUNCTION_ARGS)
{
	int16		fc_a1 = PG_GETARG_INT16(0);
	int16		fc_a2 = PG_GETARG_INT16(1);

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(fc_a1 <= fc_a2);

	PG_RETURN_FLOAT8((double) fc_a2 - (double) fc_a1);
}

/*
 * 计算两个 int4 值之间的距离（普通减法）。
 */
Datum brin_minmax_multi_distance_int4(PG_FUNCTION_ARGS)
{
	int32		fc_a1 = PG_GETARG_INT32(0);
	int32		fc_a2 = PG_GETARG_INT32(1);

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(fc_a1 <= fc_a2);

	PG_RETURN_FLOAT8((double) fc_a2 - (double) fc_a1);
}

/*
 * 计算两个 int8 值之间的距离（普通减法）。
 */
Datum brin_minmax_multi_distance_int8(PG_FUNCTION_ARGS)
{
	int64		fc_a1 = PG_GETARG_INT64(0);
	int64		fc_a2 = PG_GETARG_INT64(1);

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(fc_a1 <= fc_a2);

	PG_RETURN_FLOAT8((double) fc_a2 - (double) fc_a1);
}

/*
 * 计算两个 tid 值之间的距离（通过将它们映射到 float8 然后相减）。
 */
Datum brin_minmax_multi_distance_tid(PG_FUNCTION_ARGS)
{
	double		fc_da1,
				fc_da2;

	ItemPointer fc_pa1 = (ItemPointer) PG_GETARG_DATUM(0);
	ItemPointer fc_pa2 = (ItemPointer) PG_GETARG_DATUM(1);

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(ItemPointerCompare(fc_pa1, fc_pa2) <= 0);

	/*
	 * 我们在这里使用不检查的变体，因为用户提供的值可能
	 * 有 (ip_posid == 0)。参见 ItemPointerCompare。
	 */
	fc_da1 = ItemPointerGetBlockNumberNoCheck(fc_pa1) * MaxHeapTuplesPerPage +
		ItemPointerGetOffsetNumberNoCheck(fc_pa1);

	fc_da2 = ItemPointerGetBlockNumberNoCheck(fc_pa2) * MaxHeapTuplesPerPage +
		ItemPointerGetOffsetNumberNoCheck(fc_pa2);

	PG_RETURN_FLOAT8(fc_da2 - fc_da1);
}

/*
 * 计算两个 numeric 值之间的距离（普通减法）。
 */
Datum brin_minmax_multi_distance_numeric(PG_FUNCTION_ARGS)
{
	Datum		fc_d;
	Datum		fc_a1 = PG_GETARG_DATUM(0);
	Datum		fc_a2 = PG_GETARG_DATUM(1);

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(DatumGetBool(DirectFunctionCall2(numeric_le, fc_a1, fc_a2)));

	fc_d = DirectFunctionCall2(numeric_sub, fc_a2, fc_a1);	/* a2 - a1 */

	PG_RETURN_FLOAT8(DirectFunctionCall1(numeric_float8, fc_d));
}

/*
 * 计算两个 UUID 值之间的近似距离。
 *
 * XXX 我们不需要一个完全准确的值，因此我们用一个 64 位浮点数
 * 来近似 delta（这应该是 128 位整数）。
 * 小的误差在实践中并不重要，最坏情况下
 * 我们会决定合并不最接近的范围。
 */
Datum brin_minmax_multi_distance_uuid(PG_FUNCTION_ARGS)
{
	int			fc_i;
	float8		fc_delta = 0;

	Datum		fc_a1 = PG_GETARG_DATUM(0);
	Datum		fc_a2 = PG_GETARG_DATUM(1);

	pg_uuid_t  *fc_u1 = DatumGetUUIDP(fc_a1);
	pg_uuid_t  *fc_u2 = DatumGetUUIDP(fc_a2);

	/*
	 * 我们知道这些值是范围边界，但范围可能会被压缩
	 * （即单个点），具有相等的值。
	 */
	Assert(DatumGetBool(DirectFunctionCall2(uuid_le, fc_a1, fc_a2)));

	/* 计算近似 delta 作为双精度值 */
	for (fc_i = UUID_LEN - 1; fc_i >= 0; fc_i--)
	{
		fc_delta += (int) fc_u2->data[fc_i] - (int) fc_u1->data[fc_i];
		fc_delta /= 256;
	}

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个日期之间的近似距离。
 */
Datum brin_minmax_multi_distance_date(PG_FUNCTION_ARGS)
{
	float8		fc_delta = 0;
	DateADT		fc_dateVal1 = PG_GETARG_DATEADT(0);
	DateADT		fc_dateVal2 = PG_GETARG_DATEADT(1);

	fc_delta = (float8) fc_dateVal2 - (float8) fc_dateVal1;

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个时间（不带时区）值之间的近似距离。
 *
 * TimeADT 只是一个 int64，因此我们直接相减这些值。
 */
Datum brin_minmax_multi_distance_time(PG_FUNCTION_ARGS)
{
	float8		fc_delta = 0;

	TimeADT		fc_ta = PG_GETARG_TIMEADT(0);
	TimeADT		fc_tb = PG_GETARG_TIMEADT(1);

	fc_delta = (fc_tb - fc_ta);

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个 timetz 值之间的近似距离。
 *
 * 直接减去嵌入在 TimeTzADT 中的 TimeADT（int64）值。
 */
Datum brin_minmax_multi_distance_timetz(PG_FUNCTION_ARGS)
{
	float8		fc_delta = 0;

	TimeTzADT  *fc_ta = PG_GETARG_TIMETZADT_P(0);
	TimeTzADT  *fc_tb = PG_GETARG_TIMETZADT_P(1);

	fc_delta = (fc_tb->time - fc_ta->time) + (fc_tb->zone - fc_ta->zone) * USECS_PER_SEC;

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个时间戳值之间的距离。
 */
Datum brin_minmax_multi_distance_timestamp(PG_FUNCTION_ARGS)
{
	float8		fc_delta = 0;

	Timestamp	fc_dt1 = PG_GETARG_TIMESTAMP(0);
	Timestamp	fc_dt2 = PG_GETARG_TIMESTAMP(1);

	fc_delta = (float8) fc_dt2 - (float8) fc_dt1;

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个间隔值之间的距离。
 */
Datum brin_minmax_multi_distance_interval(PG_FUNCTION_ARGS)
{
	float8		fc_delta = 0;

	Interval   *fc_ia = PG_GETARG_INTERVAL_P(0);
	Interval   *fc_ib = PG_GETARG_INTERVAL_P(1);

	int64		fc_dayfraction;
	int64		fc_days;

	/*
	 * delta是两个时间间隔之间的（分数）天数。假设
	 * 月份有30天，以便与interval_cmp_internal保持一致。我们
	 * 不需要精确，在最坏的情况下我们会构建效率稍低的范围。
	 * 但是我们不应该与interval_cmp相矛盾。
	 */
	fc_dayfraction = (fc_ib->time % USECS_PER_DAY) - (fc_ia->time % USECS_PER_DAY);
	fc_days = (fc_ib->time / USECS_PER_DAY) - (fc_ia->time / USECS_PER_DAY);
	fc_days += (int64) fc_ib->day - (int64) fc_ia->day;
	fc_days += ((int64) fc_ib->month - (int64) fc_ia->month) * INT64CONST(30);

	/* 转换为双精度 */
	fc_delta = (double) fc_days + fc_dayfraction / (double) USECS_PER_DAY;

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个pg_lsn值之间的距离。
 *
 * LSN只是流中的一个int64编码位置，因此直接减去
 * 这些int64值。
 */
Datum brin_minmax_multi_distance_pg_lsn(PG_FUNCTION_ARGS)
{
	float8		fc_delta = 0;

	XLogRecPtr	fc_lsna = PG_GETARG_LSN(0);
	XLogRecPtr	fc_lsnb = PG_GETARG_LSN(1);

	fc_delta = (fc_lsnb - fc_lsna);

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个macaddr值之间的距离。
 *
 * mac地址被视为6个无符号字符，因此对UUID值
 * 进行相同的处理。
 */
Datum brin_minmax_multi_distance_macaddr(PG_FUNCTION_ARGS)
{
	float8		fc_delta;

	macaddr    *fc_a = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_b = PG_GETARG_MACADDR_P(1);

	fc_delta = ((float8) fc_b->f - (float8) fc_a->f);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->e - (float8) fc_a->e);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->d - (float8) fc_a->d);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->c - (float8) fc_a->c);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->b - (float8) fc_a->b);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->a - (float8) fc_a->a);
	fc_delta /= 256;

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个macaddr8值之间的距离。
 *
 * macaddr8地址是8个无符号字符，因此对UUID值
 * 进行相同的处理。
 */
Datum brin_minmax_multi_distance_macaddr8(PG_FUNCTION_ARGS)
{
	float8		fc_delta;

	macaddr8   *fc_a = PG_GETARG_MACADDR8_P(0);
	macaddr8   *fc_b = PG_GETARG_MACADDR8_P(1);

	fc_delta = ((float8) fc_b->h - (float8) fc_a->h);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->g - (float8) fc_a->g);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->f - (float8) fc_a->f);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->e - (float8) fc_a->e);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->d - (float8) fc_a->d);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->c - (float8) fc_a->c);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->b - (float8) fc_a->b);
	fc_delta /= 256;

	fc_delta += ((float8) fc_b->a - (float8) fc_a->a);
	fc_delta /= 256;

	Assert(fc_delta >= 0);

	PG_RETURN_FLOAT8(fc_delta);
}

/*
 * 计算两个inet值之间的距离。
 *
 * 距离定义为32位/128位值之间的差，具体取决于IP版本。
 * 距离通过减去字节并将其规范化到每个IP族的[0,1]区间来计算。
 * 来自不同族的地址被视为在最大距离内，即1.0。
 *
 * XXX 这是否需要考虑掩码（位）？目前被忽略。
 */
Datum brin_minmax_multi_distance_inet(PG_FUNCTION_ARGS)
{
	float8		fc_delta;
	int			fc_i;
	int			fc_len;
	unsigned char *fc_addra,
			   *fc_addrb;

	inet	   *fc_ipa = PG_GETARG_INET_PP(0);
	inet	   *fc_ipb = PG_GETARG_INET_PP(1);

	int			fc_lena,
				fc_lenb;

	/*
	 * 如果地址来自不同的族，认为它们之间的距离为
	 * 最大可能距离（即1.0）。
	 */
	if (ip_family(fc_ipa) != ip_family(fc_ipb))
		PG_RETURN_FLOAT8(1.0);

	fc_addra = (unsigned char *) palloc(ip_addrsize(fc_ipa));
	memcpy(fc_addra, ip_addr(fc_ipa), ip_addrsize(fc_ipa));

	fc_addrb = (unsigned char *) palloc(ip_addrsize(fc_ipb));
	memcpy(fc_addrb, ip_addr(fc_ipb), ip_addrsize(fc_ipb));

	/*
	 * 长度是从掩码长度计算的，因为我们按范围的
	 * 第一个地址对地址进行排序，因此A.B.C.D/24 < A.B.C.1（
	 * 第一个范围从A.B.C.0开始，位于A.B.C.1之前）。在这种情况下我们
	 * 不想产生负的delta，因此我们只需去掉多余的
	 * 字节。
	 *
	 * XXX 也许这应该更加小心，去掉位，而不仅仅是
	 * 整个字节。
	 */
	fc_lena = ip_bits(fc_ipa);
	fc_lenb = ip_bits(fc_ipb);

	fc_len = ip_addrsize(fc_ipa);

	/* 将网络掩码应用于两个地址 */
	for (fc_i = 0; fc_i < fc_len; fc_i++)
	{
		unsigned char fc_mask;
		int			fc_nbits;

		fc_nbits = Max(0, fc_lena - (fc_i * 8));
		if (fc_nbits < 8)
		{
			fc_mask = (0xFF << (8 - fc_nbits));
			fc_addra[fc_i] = (fc_addra[fc_i] & fc_mask);
		}

		fc_nbits = Max(0, fc_lenb - (fc_i * 8));
		if (fc_nbits < 8)
		{
			fc_mask = (0xFF << (8 - fc_nbits));
			fc_addrb[fc_i] = (fc_addrb[fc_i] & fc_mask);
		}
	}

	/* 计算地址之间的差异。 */
	fc_delta = 0;
	for (fc_i = fc_len - 1; fc_i >= 0; fc_i--)
	{
		unsigned char fc_a = fc_addra[fc_i];
		unsigned char fc_b = fc_addrb[fc_i];

		fc_delta += (float8) fc_b - (float8) fc_a;
		fc_delta /= 256;
	}

	Assert((fc_delta >= 0) && (fc_delta <= 1));

	pfree(fc_addra);
	pfree(fc_addrb);

	PG_RETURN_FLOAT8(fc_delta);
}

static void fc_brin_minmax_multi_serialize(BrinDesc *fc_bdesc, Datum fc_src, Datum *fc_dst)
{
	Ranges	   *fc_ranges = (Ranges *) DatumGetPointer(fc_src);
	SerializedRanges *fc_s;

	/*
	 * 在批处理模式下，我们需要将累积的值压缩到
	 * 实际请求的值/范围的数量。
	 */
	fc_compactify_ranges(fc_bdesc, fc_ranges, fc_ranges->target_maxvalues);

	/* 此时一切都必须完全排序。 */
	Assert(fc_ranges->nsorted == fc_ranges->nvalues);

	fc_s = fc_brin_range_serialize(fc_ranges);
	fc_dst[0] = PointerGetDatum(fc_s);
}

static int fc_brin_minmax_multi_get_values(BrinDesc *fc_bdesc, MinMaxMultiOptions *fc_opts)
{
	return MinMaxMultiGetValuesPerRange(fc_opts);
}

/*
 * 通过将给定索引元组（包含某个页面范围的部分状态）
 * 与来自其他堆元组的给定值进行比较，以检查它。
 * 如果新值超出了现有元组值指定的最小/最大范围，
 * 更新索引元组并返回true。否则，返回false并在这种情况下
 * 不进行修改。
 */
Datum brin_minmax_multi_add_value(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_column = (BrinValues *) PG_GETARG_POINTER(1);
	Datum		fc_newval = PG_GETARG_DATUM(2);
	bool		fc_isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_DATUM(3);
	MinMaxMultiOptions *fc_opts = (MinMaxMultiOptions *) PG_GET_OPCLASS_OPTIONS();
	Oid			fc_colloid = PG_GET_COLLATION();
	bool		fc_modified = false;
	Form_pg_attribute fc_attr;
	AttrNumber	fc_attno;
	Ranges	   *fc_ranges;
	SerializedRanges *fc_serialized = NULL;

	Assert(!fc_isnull);

	fc_attno = fc_column->bv_attno;
	fc_attr = TupleDescAttr(fc_bdesc->bd_tupdesc, fc_attno - 1);

	/* 如果可能，使用已经反序列化的值 */
	fc_ranges = (Ranges *) DatumGetPointer(fc_column->bv_mem_value);

	/*
	 * 如果这是第一个非空值，我们需要初始化范围
	 * 列表。否则，只需从 BrinValues 中提取现有范围列表。
	 *
	 * 当从空范围开始时，我们假设这是批处理模式，并且
	 * 我们使用更大的缓冲区。缓冲区大小源自 BRIN 范围
	 * 大小，每页行数，以及一些合理的最小/最大值。小缓冲区对性能
	 * 不利，但大缓冲区可能需要大量内存（因为要保留所有值）。
	 */
	if (fc_column->bv_allnulls)
	{
		MemoryContext fc_oldctx;

		int			fc_target_maxvalues;
		int			fc_maxvalues;
		BlockNumber fc_pagesPerRange = BrinGetPagesPerRange(fc_bdesc->bd_index);

		/* 指定的 reloption 是什么？ */
		fc_target_maxvalues = fc_brin_minmax_multi_get_values(fc_bdesc, fc_opts);

		/*
		 * 确定插入缓冲区大小 - 我们使用目标的10倍，限制为
		 * 堆范围内最大值的数量。这已经超过了
		 * 足够的，考虑到每页的实际行数可能
		 * 低得多，但无所谓。
		 */
		fc_maxvalues = Min(fc_target_maxvalues * MINMAX_BUFFER_FACTOR,
						MaxHeapTuplesPerPage * fc_pagesPerRange);

		/* 但始终至少是原始值 */
		fc_maxvalues = Max(fc_maxvalues, fc_target_maxvalues);

		/* 始终受 MIN/MAX 限制 */
		fc_maxvalues = Max(fc_maxvalues, MINMAX_BUFFER_MIN);
		fc_maxvalues = Min(fc_maxvalues, MINMAX_BUFFER_MAX);

		fc_oldctx = MemoryContextSwitchTo(fc_column->bv_context);
		fc_ranges = fc_minmax_multi_init(fc_maxvalues);
		fc_ranges->attno = fc_attno;
		fc_ranges->colloid = fc_colloid;
		fc_ranges->typid = fc_attr->atttypid;
		fc_ranges->target_maxvalues = fc_target_maxvalues;

		/* 我们肯定需要比较器，所以现在就查找它 */
		fc_ranges->cmp = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
														 BTLessStrategyNumber);

		MemoryContextSwitchTo(fc_oldctx);

		fc_column->bv_allnulls = false;
		fc_modified = true;

		fc_column->bv_mem_value = PointerGetDatum(fc_ranges);
		fc_column->bv_serialize = fc_brin_minmax_multi_serialize;
	}
	else if (!fc_ranges)
	{
		MemoryContext fc_oldctx;

		int			fc_maxvalues;
		BlockNumber fc_pagesPerRange = BrinGetPagesPerRange(fc_bdesc->bd_index);

		fc_oldctx = MemoryContextSwitchTo(fc_column->bv_context);

		fc_serialized = (SerializedRanges *) PG_DETOAST_DATUM(fc_column->bv_values[0]);

		/*
		 * 确定插入缓冲区大小 - 我们使用目标的10倍，限制为
		 * 堆范围内最大值的数量。这已经超过了
		 * 足够的，考虑到每页的实际行数可能
		 * 低得多，但无所谓。
		 */
		fc_maxvalues = Min(fc_serialized->maxvalues * MINMAX_BUFFER_FACTOR,
						MaxHeapTuplesPerPage * fc_pagesPerRange);

		/* 但始终至少是原始值 */
		fc_maxvalues = Max(fc_maxvalues, fc_serialized->maxvalues);

		/* 始终受 MIN/MAX 限制 */
		fc_maxvalues = Max(fc_maxvalues, MINMAX_BUFFER_MIN);
		fc_maxvalues = Min(fc_maxvalues, MINMAX_BUFFER_MAX);

		fc_ranges = fc_brin_range_deserialize(fc_maxvalues, fc_serialized);

		fc_ranges->attno = fc_attno;
		fc_ranges->colloid = fc_colloid;
		fc_ranges->typid = fc_attr->atttypid;

		/* 我们肯定需要比较器，所以现在就查找它 */
		fc_ranges->cmp = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
														 BTLessStrategyNumber);

		fc_column->bv_mem_value = PointerGetDatum(fc_ranges);
		fc_column->bv_serialize = fc_brin_minmax_multi_serialize;

		MemoryContextSwitchTo(fc_oldctx);
	}

	/*
	 * 尝试将新值添加到范围中。我们需要更新修改
	 * 标志，以便稍后序列化更新的摘要。
	 */
	fc_modified |= fc_range_add_value(fc_bdesc, fc_colloid, fc_attno, fc_attr, fc_ranges, fc_newval);


	PG_RETURN_BOOL(fc_modified);
}

/*
 * 给定一个与某个页面范围相对应的索引元组和扫描键，
 * 返回扫描键是否与索引元组的最小/最大
 * 值一致。如果是，则返回 true，否则返回 false。
 */
Datum brin_minmax_multi_consistent(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_column = (BrinValues *) PG_GETARG_POINTER(1);
	ScanKey    *fc_keys = (ScanKey *) PG_GETARG_POINTER(2);
	int			fc_nkeys = PG_GETARG_INT32(3);

	Oid			fc_colloid = PG_GET_COLLATION(),
				fc_subtype;
	AttrNumber	fc_attno;
	Datum		fc_value;
	FmgrInfo   *fc_finfo;
	SerializedRanges *fc_serialized;
	Ranges	   *fc_ranges;
	int			fc_keyno;
	int			fc_rangeno;
	int			fc_i;

	fc_attno = fc_column->bv_attno;

	fc_serialized = (SerializedRanges *) PG_DETOAST_DATUM(fc_column->bv_values[0]);
	fc_ranges = fc_brin_range_deserialize(fc_serialized->maxvalues, fc_serialized);

	/* 检查范围，并对每个范围评估扫描键 */
	for (fc_rangeno = 0; fc_rangeno < fc_ranges->nranges; fc_rangeno++)
	{
		Datum		fc_minval = fc_ranges->values[2 * fc_rangeno];
		Datum		fc_maxval = fc_ranges->values[2 * fc_rangeno + 1];

		/* 假设范围匹配，我们将尝试证明否则 */
		bool		fc_matching = true;

		for (fc_keyno = 0; fc_keyno < fc_nkeys; fc_keyno++)
		{
			Datum		fc_matches;
			ScanKey		fc_key = fc_keys[fc_keyno];

			/* NULL键在bringetbitmap中处理并被过滤 */
			Assert(!(fc_key->sk_flags & SK_ISNULL));

			fc_attno = fc_key->sk_attno;
			fc_subtype = fc_key->sk_subtype;
			fc_value = fc_key->sk_argument;
			switch (fc_key->sk_strategy)
			{
				case BTLessStrategyNumber:
				case BTLessEqualStrategyNumber:
					fc_finfo = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
															   fc_key->sk_strategy);
					/* 数组中的第一个值 */
					fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_minval, fc_value);
					break;

				case BTEqualStrategyNumber:
					{
						Datum		fc_compar;
						FmgrInfo   *fc_cmpFn;

						/* 默认情况下，此范围不匹配 */
						fc_matches = false;

						/*
						 * 否则，需要将新值与
						 * 所有范围的边界进行比较。首先检查它是否
						 * 小于绝对最小值，即数组中的第一个
						 * 值。
						 */
						fc_cmpFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
																   BTGreaterStrategyNumber);
						fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_minval, fc_value);

						/* 小于此范围内的最小值 */
						if (DatumGetBool(fc_compar))
							break;

						fc_cmpFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
																   BTLessStrategyNumber);
						fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_maxval, fc_value);

						/* 大于此范围内的最大值 */
						if (DatumGetBool(fc_compar))
							break;

						/*
						 * 我们还没有设法消除此范围，因此
						 * 认为它匹配。
						 */
						fc_matches = true;

						break;
					}
				case BTGreaterEqualStrategyNumber:
				case BTGreaterStrategyNumber:
					fc_finfo = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
															   fc_key->sk_strategy);
					/* 数组中的最后一个值 */
					fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_maxval, fc_value);
					break;

				default:
					/* 不应该发生 */
					elog(ERROR, "invalid strategy number %d", fc_key->sk_strategy);
					fc_matches = 0;
					break;
			}

			/* 范围必须匹配所有扫描键 */
			fc_matching &= DatumGetBool(fc_matches);

			/* 一旦找到不匹配的键，我们就结束了 */
			if (!fc_matching)
				break;
		}

		/*
		 * 我们找到匹配所有扫描键的范围了吗？如果是，我们就完成了
		 */
		if (fc_matching)
			PG_RETURN_DATUM(BoolGetDatum(true));
	}

	/*
	 * 现在检查值。我们不在这里进行二分搜索，
	 * 因为我们处理的是序列化/完全压缩的范围，
	 * 所以应该只有很少的值。
	 */
	for (fc_i = 0; fc_i < fc_ranges->nvalues; fc_i++)
	{
		Datum		fc_val = fc_ranges->values[2 * fc_ranges->nranges + fc_i];

		/* 假设范围匹配，我们将尝试证明否则 */
		bool		fc_matching = true;

		for (fc_keyno = 0; fc_keyno < fc_nkeys; fc_keyno++)
		{
			Datum		fc_matches;
			ScanKey		fc_key = fc_keys[fc_keyno];

			/* 我们已经在开始时处理了 NULL 键 */
			if (fc_key->sk_flags & SK_ISNULL)
				continue;

			fc_attno = fc_key->sk_attno;
			fc_subtype = fc_key->sk_subtype;
			fc_value = fc_key->sk_argument;
			switch (fc_key->sk_strategy)
			{
				case BTLessStrategyNumber:
				case BTLessEqualStrategyNumber:
				case BTEqualStrategyNumber:
				case BTGreaterEqualStrategyNumber:
				case BTGreaterStrategyNumber:

					fc_finfo = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
															   fc_key->sk_strategy);
					fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_val, fc_value);
					break;

				default:
					/* 不应该发生 */
					elog(ERROR, "invalid strategy number %d", fc_key->sk_strategy);
					fc_matches = 0;
					break;
			}

			/* 范围必须匹配所有扫描键 */
			fc_matching &= DatumGetBool(fc_matches);

			/* 一旦找到不匹配的键，我们就结束了 */
			if (!fc_matching)
				break;
		}

		/* 我们找到匹配所有扫描键的范围了吗？如果是，我们就完成了 */
		if (fc_matching)
			PG_RETURN_DATUM(BoolGetDatum(true));
	}

	PG_RETURN_DATUM(BoolGetDatum(false));
}

/*
 * 给定两个 BrinValues，将第一个值更新为包含两个值中的汇总值的
 * 并集。第二个值保持不变。
 */
Datum brin_minmax_multi_union(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_col_a = (BrinValues *) PG_GETARG_POINTER(1);
	BrinValues *fc_col_b = (BrinValues *) PG_GETARG_POINTER(2);

	Oid			fc_colloid = PG_GET_COLLATION();
	SerializedRanges *fc_serialized_a;
	SerializedRanges *fc_serialized_b;
	Ranges	   *fc_ranges_a;
	Ranges	   *fc_ranges_b;
	AttrNumber	fc_attno;
	Form_pg_attribute fc_attr;
	ExpandedRange *fc_eranges;
	int			fc_neranges;
	FmgrInfo   *fc_cmpFn,
			   *fc_distanceFn;
	DistanceValue *fc_distances;
	MemoryContext fc_ctx;
	MemoryContext fc_oldctx;

	Assert(fc_col_a->bv_attno == fc_col_b->bv_attno);
	Assert(!fc_col_a->bv_allnulls && !fc_col_b->bv_allnulls);

	fc_attno = fc_col_a->bv_attno;
	fc_attr = TupleDescAttr(fc_bdesc->bd_tupdesc, fc_attno - 1);

	fc_serialized_a = (SerializedRanges *) PG_DETOAST_DATUM(fc_col_a->bv_values[0]);
	fc_serialized_b = (SerializedRanges *) PG_DETOAST_DATUM(fc_col_b->bv_values[0]);

	fc_ranges_a = fc_brin_range_deserialize(fc_serialized_a->maxvalues, fc_serialized_a);
	fc_ranges_b = fc_brin_range_deserialize(fc_serialized_b->maxvalues, fc_serialized_b);

	/* 确保两个范围都不是 NULL */
	Assert(fc_ranges_a && fc_ranges_b);

	fc_neranges = (fc_ranges_a->nranges + fc_ranges_a->nvalues) +
		(fc_ranges_b->nranges + fc_ranges_b->nvalues);

	/*
	 * distanceFn 调用（可能会内部调用 e.g. numeric_le）可能
	 * 会分配相当多的内存，我们必须避免泄漏。否则，
	 * 我们在构建索引时会遇到问题。因此我们创建一个
	 * 本地内存上下文，并确保在离开这个函数之前释放内存
	 * （而不是在每次调用后）。
	 */
	fc_ctx = AllocSetContextCreate(CurrentMemoryContext,
								"minmax-multi context",
								ALLOCSET_DEFAULT_SIZES);

	fc_oldctx = MemoryContextSwitchTo(fc_ctx);

	/* 分配并填充 */
	fc_eranges = (ExpandedRange *) palloc0(fc_neranges * sizeof(ExpandedRange));

	/* 用第一个范围的条目填充扩展范围 */
	fc_fill_expanded_ranges(fc_eranges, fc_ranges_a->nranges + fc_ranges_a->nvalues,
						 fc_ranges_a);

	/* 现在为第二个范围添加组合范围 */
	fc_fill_expanded_ranges(&fc_eranges[fc_ranges_a->nranges + fc_ranges_a->nvalues],
						 fc_ranges_b->nranges + fc_ranges_b->nvalues,
						 fc_ranges_b);

	fc_cmpFn = fc_minmax_multi_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
											   BTLessStrategyNumber);

	/* 对扩展范围进行排序 */
	fc_neranges = fc_sort_expanded_ranges(fc_cmpFn, fc_colloid, fc_eranges, fc_neranges);

	/*
	 * 我们加载了两个不同的扩展范围列表，因此它们中的某些可能
	 * 是重叠的。因此遍历它们并合并它们。
	 */
	fc_neranges = fc_merge_overlapping_ranges(fc_cmpFn, fc_colloid, fc_eranges, fc_neranges);

	/* 检查组合范围是否正确（无重叠，排序） */
	fc_AssertCheckExpandedRanges(fc_bdesc, fc_colloid, fc_attno, fc_attr, fc_eranges, fc_neranges);

	/*
	 * 如有必要，减少一些范围。
	 *
	 * XXX 这可能相当昂贵，因此我们可能应该仅在
	 * 实际需要时（当我们有太多范围时）这样做。
	 */

	/* 构建间隙距离数组并按升序排序 */
	fc_distanceFn = fc_minmax_multi_get_procinfo(fc_bdesc, fc_attno, PROCNUM_DISTANCE);
	fc_distances = fc_build_distances(fc_distanceFn, fc_colloid, fc_eranges, fc_neranges);

	/*
	 * 查看存储当前范围需要多少值，并根据需要合并尽可能多的值，
	 * 以低于阈值。折叠的范围将作为单个值存储。
	 *
	 * XXX 这不应用负载因子，因为我们不期望向范围添加更多
	 * 值，因此我们更喜欢保持尽可能多的范围。
	 *
	 * XXX 两个范围中的最大值可以不同吗？也许我们应该
	 * 使用其中的最大值？
	 */
	fc_neranges = fc_reduce_expanded_ranges(fc_eranges, fc_neranges, fc_distances,
									  fc_ranges_a->maxvalues,
									  fc_cmpFn, fc_colloid);

	/* 更新第一个范围摘要 */
	fc_store_expanded_ranges(fc_ranges_a, fc_eranges, fc_neranges);

	MemoryContextSwitchTo(fc_oldctx);
	MemoryContextDelete(fc_ctx);

	/* 清理并更新序列化值 */
	pfree(fc_serialized_a);
	fc_col_a->bv_values[0] = PointerGetDatum(fc_brin_range_serialize(fc_ranges_a));

	PG_RETURN_VOID();
}

/*
 * 缓存并返回 minmax 多操作类支持过程
 *
 * 返回与给定函数支持编号对应的过程
 * 或 null 如果不存在。
 */
static FmgrInfo * fc_minmax_multi_get_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno, uint16 fc_procnum)
{
	MinmaxMultiOpaque *fc_opaque;
	uint16		fc_basenum = fc_procnum - PROCNUM_BASE;

	/*
	 * 我们将其缓存于不透明结构中，以避免重复的系统缓存查找。
	 */
	fc_opaque = (MinmaxMultiOpaque *) fc_bdesc->bd_info[fc_attno - 1]->oi_opaque;

	/*
	 * 如果我们已经搜索过这个过程且未找到它，就不必再搜索了。
	 */
	if (fc_opaque->extra_proc_missing[fc_basenum])
		return NULL;

	if (fc_opaque->extra_procinfos[fc_basenum].fn_oid == InvalidOid)
	{
		if (RegProcedureIsValid(index_getprocid(fc_bdesc->bd_index, fc_attno,
												fc_procnum)))
		{
			fmgr_info_copy(&fc_opaque->extra_procinfos[fc_basenum],
						   index_getprocinfo(fc_bdesc->bd_index, fc_attno, fc_procnum),
						   fc_bdesc->bd_context);
		}
		else
		{
			fc_opaque->extra_proc_missing[fc_basenum] = true;
			return NULL;
		}
	}

	return &fc_opaque->extra_procinfos[fc_basenum];
}

/*
 * 缓存并返回给定策略的过程。
 *
 * 注意：此函数镜像 minmax_multi_get_strategy_procinfo；请查看
 * 其中的说明。如果在这里进行更改，请查看该函数。
 */
static FmgrInfo * fc_minmax_multi_get_strategy_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno, Oid fc_subtype,
								   uint16 fc_strategynum)
{
	MinmaxMultiOpaque *fc_opaque;

	Assert(fc_strategynum >= 1 &&
		   fc_strategynum <= BTMaxStrategyNumber);

	fc_opaque = (MinmaxMultiOpaque *) fc_bdesc->bd_info[fc_attno - 1]->oi_opaque;

	/*
	 * 我们在 opaque 结构中缓存之前子类型的过程，以避免重复的 syscache
	 * 查找。如果子类型发生更改，则使所有缓存条目无效。
	 */
	if (fc_opaque->cached_subtype != fc_subtype)
	{
		uint16		fc_i;

		for (fc_i = 1; fc_i <= BTMaxStrategyNumber; fc_i++)
			fc_opaque->strategy_procinfos[fc_i - 1].fn_oid = InvalidOid;
		fc_opaque->cached_subtype = fc_subtype;
	}

	if (fc_opaque->strategy_procinfos[fc_strategynum - 1].fn_oid == InvalidOid)
	{
		Form_pg_attribute fc_attr;
		HeapTuple	fc_tuple;
		Oid			fc_opfamily,
					fc_oprid;
		bool		fc_isNull;

		fc_opfamily = fc_bdesc->bd_index->rd_opfamily[fc_attno - 1];
		fc_attr = TupleDescAttr(fc_bdesc->bd_tupdesc, fc_attno - 1);
		fc_tuple = SearchSysCache4(AMOPSTRATEGY, ObjectIdGetDatum(fc_opfamily),
								ObjectIdGetDatum(fc_attr->atttypid),
								ObjectIdGetDatum(fc_subtype),
								Int16GetDatum(fc_strategynum));
		if (!HeapTupleIsValid(fc_tuple))
			elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
				 fc_strategynum, fc_attr->atttypid, fc_subtype, fc_opfamily);

		fc_oprid = DatumGetObjectId(SysCacheGetAttr(AMOPSTRATEGY, fc_tuple,
												 Anum_pg_amop_amopopr, &fc_isNull));
		ReleaseSysCache(fc_tuple);
		Assert(!fc_isNull && RegProcedureIsValid(fc_oprid));

		fmgr_info_cxt(get_opcode(fc_oprid),
					  &fc_opaque->strategy_procinfos[fc_strategynum - 1],
					  fc_bdesc->bd_context);
	}

	return &fc_opaque->strategy_procinfos[fc_strategynum - 1];
}

Datum brin_minmax_multi_options(PG_FUNCTION_ARGS)
{
	local_relopts *fc_relopts = (local_relopts *) PG_GETARG_POINTER(0);

	init_local_reloptions(fc_relopts, sizeof(MinMaxMultiOptions));

	add_local_int_reloption(fc_relopts, "values_per_range", "desc",
							MINMAX_MULTI_DEFAULT_VALUES_PER_PAGE, 8, 256,
							offsetof(MinMaxMultiOptions, valuesPerRange));

	PG_RETURN_VOID();
}

/*
 * brin_minmax_multi_summary_in
 *		- brin_minmax_multi_summary类型的输入例程。
 *
 * brin_minmax_multi_summary仅在内部用于表示BRIN minmax-multi索引中的摘要，因此它没有自己的操作，我们也不允许输入。
 */
Datum brin_minmax_multi_summary_in(PG_FUNCTION_ARGS)
{
	/*
	 * brin_minmax_multi_summary以二进制形式存储数据，因此不需要解析文本输入，因此不允许这样做。
	 */
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("cannot accept a value of type %s", "brin_minmax_multi_summary")));

	PG_RETURN_VOID();			/* 保持编译器安静 */
}


/*
 * brin_minmax_multi_summary_out
 *		- brin_minmax_multi_summary类型的输出例程。
 *
 * BRIN minmax-multi摘要被序列化为bytea值，但我们希望输出一些人类可以理解的更好内容。
 */
Datum brin_minmax_multi_summary_out(PG_FUNCTION_ARGS)
{
	int			fc_i;
	int			fc_idx;
	SerializedRanges *fc_ranges;
	Ranges	   *fc_ranges_deserialized;
	StringInfoData fc_str;
	bool		fc_isvarlena;
	Oid			fc_outfunc;
	FmgrInfo	fc_fmgrinfo;
	ArrayBuildState *fc_astate_values = NULL;

	initStringInfo(&fc_str);
	appendStringInfoChar(&fc_str, '{');

	/*
	 * 解除压缩以获取带有完整4B头的值（不能存储在toast表中，但可以使用1B头）。
	 */
	fc_ranges = (SerializedRanges *) PG_DETOAST_DATUM(PG_GETARG_BYTEA_PP(0));

	/* 查找该类型的输出函数 */
	getTypeOutputInfo(fc_ranges->typid, &fc_outfunc, &fc_isvarlena);
	fmgr_info(fc_outfunc, &fc_fmgrinfo);

	/* 反序列化范围信息为易于处理的片段 */
	fc_ranges_deserialized = fc_brin_range_deserialize(fc_ranges->maxvalues, fc_ranges);

	appendStringInfo(&fc_str, "nranges: %d  nvalues: %d  maxvalues: %d",
					 fc_ranges_deserialized->nranges,
					 fc_ranges_deserialized->nvalues,
					 fc_ranges_deserialized->maxvalues);

	/* 序列化范围 */
	fc_idx = 0;
	for (fc_i = 0; fc_i < fc_ranges_deserialized->nranges; fc_i++)
	{
		char	   *fc_a,
				   *fc_b;
		text	   *fc_c;
		StringInfoData fc_str;

		initStringInfo(&fc_str);

		fc_a = OutputFunctionCall(&fc_fmgrinfo, fc_ranges_deserialized->values[fc_idx++]);
		fc_b = OutputFunctionCall(&fc_fmgrinfo, fc_ranges_deserialized->values[fc_idx++]);

		appendStringInfo(&fc_str, "%s ... %s", fc_a, fc_b);

		fc_c = cstring_to_text(fc_str.data);

		fc_astate_values = accumArrayResult(fc_astate_values,
										 PointerGetDatum(fc_c),
										 false,
										 TEXTOID,
										 CurrentMemoryContext);
	}

	if (fc_ranges_deserialized->nranges > 0)
	{
		Oid			fc_typoutput;
		bool		fc_typIsVarlena;
		Datum		fc_val;
		char	   *fc_extval;

		getTypeOutputInfo(ANYARRAYOID, &fc_typoutput, &fc_typIsVarlena);

		fc_val = PointerGetDatum(makeArrayResult(fc_astate_values, CurrentMemoryContext));

		fc_extval = OidOutputFunctionCall(fc_typoutput, fc_val);

		appendStringInfo(&fc_str, " ranges: %s", fc_extval);
	}

	/* 序列化单个值 */
	fc_astate_values = NULL;

	for (fc_i = 0; fc_i < fc_ranges_deserialized->nvalues; fc_i++)
	{
		Datum		fc_a;
		text	   *fc_b;
		StringInfoData fc_str;

		initStringInfo(&fc_str);

		fc_a = FunctionCall1(&fc_fmgrinfo, fc_ranges_deserialized->values[fc_idx++]);

		appendStringInfoString(&fc_str, DatumGetCString(fc_a));

		fc_b = cstring_to_text(fc_str.data);

		fc_astate_values = accumArrayResult(fc_astate_values,
										 PointerGetDatum(fc_b),
										 false,
										 TEXTOID,
										 CurrentMemoryContext);
	}

	if (fc_ranges_deserialized->nvalues > 0)
	{
		Oid			fc_typoutput;
		bool		fc_typIsVarlena;
		Datum		fc_val;
		char	   *fc_extval;

		getTypeOutputInfo(ANYARRAYOID, &fc_typoutput, &fc_typIsVarlena);

		fc_val = PointerGetDatum(makeArrayResult(fc_astate_values, CurrentMemoryContext));

		fc_extval = OidOutputFunctionCall(fc_typoutput, fc_val);

		appendStringInfo(&fc_str, " values: %s", fc_extval);
	}


	appendStringInfoChar(&fc_str, '}');

	PG_RETURN_CSTRING(fc_str.data);
}

/*
 * brin_minmax_multi_summary_recv
 *		- brin_minmax_multi_summary类型的二进制输入例程。
 */
Datum brin_minmax_multi_summary_recv(PG_FUNCTION_ARGS)
{
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("cannot accept a value of type %s", "brin_minmax_multi_summary")));

	PG_RETURN_VOID();			/* 保持编译器安静 */
}

/*
 * brin_minmax_multi_summary_send
 *		- brin_minmax_multi_summary类型的二进制输出例程。
 *
 * BRIN minmax-multi摘要被序列化为bytea值（尽管类型命名不同），所以我们就发送这个吧。
 */
Datum brin_minmax_multi_summary_send(PG_FUNCTION_ARGS)
{
	return byteasend(fcinfo);
}
